feat(opensearch): add new check opensearch_service_domains_fault_tolerant_data_nodes (#5366)

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

prowler/providers/aws/services/opensearch/opensearch_service.py

@@ -13,142 +13,142 @@ class OpenSearchService(AWSService):
     def __init__(self, provider):
         # Call AWSService's __init__
         super().__init__("opensearch", provider)
-        self.opensearch_domains = []
+        self.opensearch_domains = {}
         self.__threading_call__(self._list_domain_names)
-        self._describe_domain_config(self.regional_clients)
-        self._describe_domain(self.regional_clients)
-        self._list_tags()
+        self.__threading_call__(
+            self._describe_domain_config, self.opensearch_domains.values()
+        )
+        self.__threading_call__(self._describe_domain, self.opensearch_domains.values())
+        self.__threading_call__(self._list_tags, self.opensearch_domains.values())
 
     def _list_domain_names(self, regional_client):
         logger.info("OpenSearch - listing domain names...")
         try:
             domains = regional_client.list_domain_names()
             for domain in domains["DomainNames"]:
-                arn = f"arn:{self.audited_partition}:opensearch:{regional_client.region}:{self.audited_account}:domain/{domain['DomainName']}"
+                arn = f"arn:{self.audited_partition}:es:{regional_client.region}:{self.audited_account}:domain/{domain['DomainName']}"
                 if not self.audit_resources or (
                     is_resource_filtered(arn, self.audit_resources)
                 ):
-                    self.opensearch_domains.append(
-                        OpenSearchDomain(
-                            arn=arn,
-                            name=domain["DomainName"],
-                            region=regional_client.region,
-                        )
+                    self.opensearch_domains[arn] = OpenSearchDomain(
+                        arn=arn,
+                        name=domain["DomainName"],
+                        region=regional_client.region,
                     )
         except Exception as error:
             logger.error(
                 f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
             )
 
-    def _describe_domain_config(self, regional_clients):
+    def _describe_domain_config(self, domain):
         logger.info("OpenSearch - describing domain configurations...")
         try:
-            for domain in self.opensearch_domains:
-                regional_client = regional_clients[domain.region]
-                describe_domain = regional_client.describe_domain_config(
-                    DomainName=domain.name
-                )
-                for logging_key in [
-                    "SEARCH_SLOW_LOGS",
-                    "INDEX_SLOW_LOGS",
-                    "AUDIT_LOGS",
-                ]:
-                    if logging_key in describe_domain["DomainConfig"].get(
-                        "LogPublishingOptions", {}
-                    ).get("Options", {}):
-                        domain.logging.append(
-                            PublishingLoggingOption(
-                                name=logging_key,
-                                enabled=describe_domain["DomainConfig"][
-                                    "LogPublishingOptions"
-                                ]["Options"][logging_key]["Enabled"],
-                            )
+            regional_client = self.regional_clients[domain.region]
+            describe_domain = regional_client.describe_domain_config(
+                DomainName=domain.name
+            )
+            for logging_key in [
+                "SEARCH_SLOW_LOGS",
+                "INDEX_SLOW_LOGS",
+                "AUDIT_LOGS",
+            ]:
+                if logging_key in describe_domain["DomainConfig"].get(
+                    "LogPublishingOptions", {}
+                ).get("Options", {}):
+                    domain.logging.append(
+                        PublishingLoggingOption(
+                            name=logging_key,
+                            enabled=describe_domain["DomainConfig"][
+                                "LogPublishingOptions"
+                            ]["Options"][logging_key]["Enabled"],
                         )
-                try:
-                    domain.access_policy = loads(
-                        describe_domain["DomainConfig"]["AccessPolicies"]["Options"]
                     )
-                except JSONDecodeError as error:
-                    logger.warning(
-                        f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
-                    )
-                    continue
-
-        except Exception as error:
-            logger.error(
-                f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
-            )
-
-    def _describe_domain(self, regional_clients):
-        logger.info("OpenSearch - describing domain configurations...")
-        try:
-            for domain in self.opensearch_domains:
-                regional_client = regional_clients[domain.region]
-                describe_domain = regional_client.describe_domain(
-                    DomainName=domain.name
-                )
-                domain.arn = describe_domain["DomainStatus"]["ARN"]
-                domain.vpc_endpoints = None
-                if "Endpoints" in describe_domain["DomainStatus"]:
-                    if "vpc" in describe_domain["DomainStatus"]["Endpoints"]:
-                        domain.vpc_endpoints = [
-                            vpc
-                            for vpc in describe_domain["DomainStatus"][
-                                "Endpoints"
-                            ].values()
-                        ]
-
-                domain.vpc_id = None
-                if "VPCOptions" in describe_domain["DomainStatus"]:
-                    domain.vpc_id = describe_domain["DomainStatus"]["VPCOptions"][
-                        "VPCId"
-                    ]
-                domain.cognito_options = describe_domain["DomainStatus"][
-                    "CognitoOptions"
-                ]["Enabled"]
-                domain.encryption_at_rest = describe_domain["DomainStatus"][
-                    "EncryptionAtRestOptions"
-                ]["Enabled"]
-                domain.node_to_node_encryption = describe_domain["DomainStatus"][
-                    "NodeToNodeEncryptionOptions"
-                ]["Enabled"]
-                domain.enforce_https = describe_domain["DomainStatus"][
-                    "DomainEndpointOptions"
-                ]["EnforceHTTPS"]
-                domain.internal_user_database = describe_domain["DomainStatus"][
-                    "AdvancedSecurityOptions"
-                ]["InternalUserDatabaseEnabled"]
-                domain.saml_enabled = (
-                    describe_domain["DomainStatus"]["AdvancedSecurityOptions"]
-                    .get("SAMLOptions", {})
-                    .get("Enabled", False)
-                )
-                domain.update_available = describe_domain["DomainStatus"][
-                    "ServiceSoftwareOptions"
-                ]["UpdateAvailable"]
-                domain.version = describe_domain["DomainStatus"]["EngineVersion"]
-                domain.advanced_settings_enabled = describe_domain["DomainStatus"][
-                    "AdvancedSecurityOptions"
-                ]["Enabled"]
-        except Exception as error:
-            logger.error(
-                f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
-            )
-
-    def _list_tags(self):
-        logger.info("OpenSearch - List Tags...")
-        for domain in self.opensearch_domains:
             try:
-                regional_client = self.regional_clients[domain.region]
-                response = regional_client.list_tags(
-                    ARN=domain.arn,
-                )["TagList"]
-                domain.tags = response
-            except Exception as error:
-                logger.error(
+                domain.access_policy = loads(
+                    describe_domain["DomainConfig"]["AccessPolicies"]["Options"]
+                )
+            except JSONDecodeError as error:
+                logger.warning(
                     f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
                 )
 
+        except Exception as error:
+            logger.error(
+                f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+            )
+
+    def _describe_domain(self, domain):
+        logger.info("OpenSearch - describing domain configurations...")
+        try:
+            regional_client = self.regional_clients[domain.region]
+            describe_domain = regional_client.describe_domain(DomainName=domain.name)
+            domain.arn = describe_domain["DomainStatus"]["ARN"]
+            domain.vpc_endpoints = None
+            if "Endpoints" in describe_domain["DomainStatus"]:
+                if "vpc" in describe_domain["DomainStatus"]["Endpoints"]:
+                    domain.vpc_endpoints = [
+                        vpc
+                        for vpc in describe_domain["DomainStatus"]["Endpoints"].values()
+                    ]
+
+            domain.vpc_id = None
+            if "VPCOptions" in describe_domain["DomainStatus"]:
+                domain.vpc_id = describe_domain["DomainStatus"]["VPCOptions"].get(
+                    "VPCId", None
+                )
+            domain.cognito_options = describe_domain["DomainStatus"][
+                "CognitoOptions"
+            ].get("Enabled", False)
+            domain.encryption_at_rest = describe_domain["DomainStatus"][
+                "EncryptionAtRestOptions"
+            ].get("Enabled", False)
+            domain.node_to_node_encryption = describe_domain["DomainStatus"][
+                "NodeToNodeEncryptionOptions"
+            ].get("Enabled", False)
+            domain.enforce_https = describe_domain["DomainStatus"][
+                "DomainEndpointOptions"
+            ].get("EnforceHTTPS", False)
+            domain.internal_user_database = describe_domain["DomainStatus"][
+                "AdvancedSecurityOptions"
+            ].get("InternalUserDatabaseEnabled", False)
+            domain.saml_enabled = (
+                describe_domain["DomainStatus"]["AdvancedSecurityOptions"]
+                .get("SAMLOptions", {})
+                .get("Enabled", False)
+            )
+            domain.update_available = (
+                describe_domain["DomainStatus"]
+                .get("ServiceSoftwareOptions", {"UpdateAvailable": False})
+                .get("UpdateAvailable", False)
+            )
+            domain.version = describe_domain["DomainStatus"].get("EngineVersion", None)
+            domain.advanced_settings_enabled = describe_domain["DomainStatus"][
+                "AdvancedSecurityOptions"
+            ].get("Enabled", False)
+            domain.instance_count = describe_domain["DomainStatus"][
+                "ClusterConfig"
+            ].get("InstanceCount", None)
+            domain.zone_awareness_enabled = describe_domain["DomainStatus"][
+                "ClusterConfig"
+            ].get("ZoneAwarenessEnabled", False)
+        except Exception as error:
+            logger.error(
+                f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+            )
+
+    def _list_tags(self, domain):
+        logger.info("OpenSearch - List Tags...")
+        try:
+            regional_client = self.regional_clients[domain.region]
+            response = regional_client.list_tags(
+                ARN=domain.arn,
+            )["TagList"]
+            domain.tags = response
+        except Exception as error:
+            logger.error(
+                f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+            )
+
 
 class PublishingLoggingOption(BaseModel):
     name: str
@@ -171,5 +171,7 @@ class OpenSearchDomain(BaseModel):
     saml_enabled: bool = None
     update_available: bool = None
     version: str = None
+    instance_count: Optional[int]
+    zone_awareness_enabled: Optional[bool]
     tags: Optional[list] = []
     advanced_settings_enabled: bool = None

prowler/providers/aws/services/opensearch/opensearch_service_domains_access_control_enabled/opensearch_service_domains_access_control_enabled.py

@@ -7,7 +7,7 @@ from prowler.providers.aws.services.opensearch.opensearch_client import (
 class opensearch_service_domains_access_control_enabled(Check):
     def execute(self):
         findings = []
-        for domain in opensearch_client.opensearch_domains:
+        for domain in opensearch_client.opensearch_domains.values():
             report = Check_Report_AWS(self.metadata())
             report.region = domain.region
             report.resource_id = domain.name

prowler/providers/aws/services/opensearch/opensearch_service_domains_audit_logging_enabled/opensearch_service_domains_audit_logging_enabled.py

@@ -7,7 +7,7 @@ from prowler.providers.aws.services.opensearch.opensearch_client import (
 class opensearch_service_domains_audit_logging_enabled(Check):
     def execute(self):
         findings = []
-        for domain in opensearch_client.opensearch_domains:
+        for domain in opensearch_client.opensearch_domains.values():
             report = Check_Report_AWS(self.metadata())
             report.region = domain.region
             report.resource_id = domain.name

prowler/providers/aws/services/opensearch/opensearch_service_domains_cloudwatch_logging_enabled/opensearch_service_domains_cloudwatch_logging_enabled.py

@@ -7,7 +7,7 @@ from prowler.providers.aws.services.opensearch.opensearch_client import (
 class opensearch_service_domains_cloudwatch_logging_enabled(Check):
     def execute(self):
         findings = []
-        for domain in opensearch_client.opensearch_domains:
+        for domain in opensearch_client.opensearch_domains.values():
             report = Check_Report_AWS(self.metadata())
             report.region = domain.region
             report.resource_id = domain.name

prowler/providers/aws/services/opensearch/opensearch_service_domains_encryption_at_rest_enabled/opensearch_service_domains_encryption_at_rest_enabled.py

@@ -7,7 +7,7 @@ from prowler.providers.aws.services.opensearch.opensearch_client import (
 class opensearch_service_domains_encryption_at_rest_enabled(Check):
     def execute(self):
         findings = []
-        for domain in opensearch_client.opensearch_domains:
+        for domain in opensearch_client.opensearch_domains.values():
             report = Check_Report_AWS(self.metadata())
             report.region = domain.region
             report.resource_id = domain.name

prowler/providers/aws/services/opensearch/opensearch_service_domains_fault_tolerant_data_nodes/opensearch_service_domains_fault_tolerant_data_nodes.metadata.json

@@ -0,0 +1,34 @@
+{
+  "Provider": "aws",
+  "CheckID": "opensearch_service_domains_fault_tolerant_data_nodes",
+  "CheckTitle": "Ensure Elasticsearch/Opensearch domains have fault-tolerant data nodes.",
+  "CheckType": [
+    "Software and Configuration Checks/AWS Security Best Practices"
+  ],
+  "ServiceName": "opensearch",
+  "SubServiceName": "",
+  "ResourceIdTemplate": "arn:partition:es:{region}:{account-id}:domain/{domain-name}",
+  "Severity": "medium",
+  "ResourceType": "AwsElasticsearchDomain",
+  "Description": "This control checks whether Elasticsearch/Opensearch domains are fault-tolerant with at least three data nodes and cross-zone replication (Zone Awareness) enabled.",
+  "Risk": "Without at least three data nodes and without cross-zone replication (Zone Awareness), the Elasticsearch/Opensearch domain may not be fault-tolerant, leading to a higher risk of data loss or unavailability in case of node failure.",
+  "RelatedUrl": "https://docs.aws.amazon.com/opensearch-service/latest/developerguide/what-is.html",
+  "Remediation": {
+    "Code": {
+      "CLI": "aws opensearch update-domain-config --domain-name <domain-name> --cluster-config InstanceCount=3,ZoneAwarenessEnabled=true",
+      "NativeIaC": "",
+      "Other": "https://docs.aws.amazon.com/securityhub/latest/userguide/es-controls.html#es-6",
+      "Terraform": ""
+    },
+    "Recommendation": {
+      "Text": "Modify the Elasticsearch/Opensearch domain to ensure at least three data nodes and enable cross-zone replication (Zone Awareness) for high availability and fault tolerance.",
+      "Url": "https://docs.aws.amazon.com/opensearch-service/latest/developerguide/managedomains-multiaz.html"
+    }
+  },
+  "Categories": [
+    "redundancy"
+  ],
+  "DependsOn": [],
+  "RelatedTo": [],
+  "Notes": ""
+}

prowler/providers/aws/services/opensearch/opensearch_service_domains_fault_tolerant_data_nodes/opensearch_service_domains_fault_tolerant_data_nodes.py

@@ -0,0 +1,33 @@
+from prowler.lib.check.models import Check, Check_Report_AWS
+from prowler.providers.aws.services.opensearch.opensearch_client import (
+    opensearch_client,
+)
+
+
+class opensearch_service_domains_fault_tolerant_data_nodes(Check):
+    def execute(self):
+        findings = []
+
+        for domain in opensearch_client.opensearch_domains.values():
+            report = Check_Report_AWS(self.metadata())
+            report.region = domain.region
+            report.resource_id = domain.name
+            report.resource_arn = domain.arn
+            report.resource_tags = domain.tags
+
+            report.status = "FAIL"
+            report.status_extended = f"Opensearch domain {domain.name} is not fault tolerant as it has less than 3 data nodes and cross-zone replication (Zone Awareness) is not enabled."
+
+            if domain.instance_count >= 3 and domain.zone_awareness_enabled:
+                report.status = "PASS"
+                report.status_extended = f"Opensearch domain {domain.name} is fault tolerant with {domain.instance_count} data nodes and cross-zone replication (Zone Awareness) enabled."
+            elif domain.instance_count >= 3 and not domain.zone_awareness_enabled:
+                report.status = "FAIL"
+                report.status_extended = f"Opensearch domain {domain.name} is not fault tolerant as it has {domain.instance_count} data nodes, but cross-zone replication (Zone Awareness) is not enabled."
+            elif domain.instance_count < 3 and domain.zone_awareness_enabled:
+                report.status = "FAIL"
+                report.status_extended = f"Opensearch domain {domain.name} is not fault tolerant as it has cross-zone replication (Zone Awareness) enabled, but only {domain.instance_count} data nodes."
+
+            findings.append(report)
+
+        return findings

prowler/providers/aws/services/opensearch/opensearch_service_domains_https_communications_enforced/opensearch_service_domains_https_communications_enforced.py

@@ -7,7 +7,7 @@ from prowler.providers.aws.services.opensearch.opensearch_client import (
 class opensearch_service_domains_https_communications_enforced(Check):
     def execute(self):
         findings = []
-        for domain in opensearch_client.opensearch_domains:
+        for domain in opensearch_client.opensearch_domains.values():
             report = Check_Report_AWS(self.metadata())
             report.region = domain.region
             report.resource_id = domain.name

prowler/providers/aws/services/opensearch/opensearch_service_domains_internal_user_database_enabled/opensearch_service_domains_internal_user_database_enabled.py

@@ -7,7 +7,7 @@ from prowler.providers.aws.services.opensearch.opensearch_client import (
 class opensearch_service_domains_internal_user_database_enabled(Check):
     def execute(self):
         findings = []
-        for domain in opensearch_client.opensearch_domains:
+        for domain in opensearch_client.opensearch_domains.values():
             report = Check_Report_AWS(self.metadata())
             report.region = domain.region
             report.resource_id = domain.name

prowler/providers/aws/services/opensearch/opensearch_service_domains_node_to_node_encryption_enabled/opensearch_service_domains_node_to_node_encryption_enabled.py

@@ -7,7 +7,7 @@ from prowler.providers.aws.services.opensearch.opensearch_client import (
 class opensearch_service_domains_node_to_node_encryption_enabled(Check):
     def execute(self):
         findings = []
-        for domain in opensearch_client.opensearch_domains:
+        for domain in opensearch_client.opensearch_domains.values():
             report = Check_Report_AWS(self.metadata())
             report.region = domain.region
             report.resource_id = domain.name

prowler/providers/aws/services/opensearch/opensearch_service_domains_not_publicly_accessible/opensearch_service_domains_not_publicly_accessible.py

@@ -7,7 +7,7 @@ from prowler.providers.aws.services.opensearch.opensearch_client import (
 class opensearch_service_domains_not_publicly_accessible(Check):
     def execute(self):
         findings = []
-        for domain in opensearch_client.opensearch_domains:
+        for domain in opensearch_client.opensearch_domains.values():
             report = Check_Report_AWS(self.metadata())
             report.region = domain.region
             report.resource_id = domain.name

prowler/providers/aws/services/opensearch/opensearch_service_domains_updated_to_the_latest_service_software_version/opensearch_service_domains_updated_to_the_latest_service_software_version.py

@@ -7,14 +7,14 @@ from prowler.providers.aws.services.opensearch.opensearch_client import (
 class opensearch_service_domains_updated_to_the_latest_service_software_version(Check):
     def execute(self):
         findings = []
-        for domain in opensearch_client.opensearch_domains:
+        for domain in opensearch_client.opensearch_domains.values():
             report = Check_Report_AWS(self.metadata())
             report.region = domain.region
             report.resource_id = domain.name
             report.resource_arn = domain.arn
             report.resource_tags = domain.tags
             report.status = "PASS"
-            report.status_extended = f"Opensearch domain {domain.name}  with version {domain.version} does not have internal updates available."
+            report.status_extended = f"Opensearch domain {domain.name} with version {domain.version} does not have internal updates available."
             if domain.update_available:
                 report.status = "FAIL"
                 report.status_extended = f"Opensearch domain {domain.name} with version {domain.version} has internal updates available."

prowler/providers/aws/services/opensearch/opensearch_service_domains_use_cognito_authentication_for_kibana/opensearch_service_domains_use_cognito_authentication_for_kibana.py

@@ -7,7 +7,7 @@ from prowler.providers.aws.services.opensearch.opensearch_client import (
 class opensearch_service_domains_use_cognito_authentication_for_kibana(Check):
     def execute(self):
         findings = []
-        for domain in opensearch_client.opensearch_domains:
+        for domain in opensearch_client.opensearch_domains.values():
             report = Check_Report_AWS(self.metadata())
             report.region = domain.region
             report.resource_id = domain.name

tests/providers/aws/services/opensearch/opensearch_service_domains_access_control_enabled/opensearch_service_domains_access_control_enabled_test.py

@@ -1,26 +1,32 @@
-from re import search
 from unittest import mock
 
-from prowler.providers.aws.services.opensearch.opensearch_service import (
-    OpenSearchDomain,
-)
-from tests.providers.aws.utils import AWS_ACCOUNT_NUMBER, AWS_REGION_EU_WEST_1
+from boto3 import client
+from moto import mock_aws
 
-domain_name = "test-domain"
-domain_arn = f"arn:aws:es:us-west-2:{AWS_ACCOUNT_NUMBER}:domain/{domain_name}"
+from tests.providers.aws.utils import (
+    AWS_ACCOUNT_NUMBER,
+    AWS_REGION_US_EAST_1,
+    set_mocked_aws_provider,
+)
 
 
 class Test_opensearch_service_domains_access_control_enabled:
+    @mock_aws
     def test_no_domains(self):
-        opensearch_client = mock.MagicMock
-        opensearch_client.opensearch_domains = []
+        client("opensearch", region_name=AWS_REGION_US_EAST_1)
+
+        from prowler.providers.aws.services.opensearch.opensearch_service import (
+            OpenSearchService,
+        )
+
+        mocked_aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
 
         with mock.patch(
-            "prowler.providers.aws.services.opensearch.opensearch_service.OpenSearchService",
-            new=opensearch_client,
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=mocked_aws_provider,
         ), mock.patch(
             "prowler.providers.aws.services.opensearch.opensearch_service_domains_access_control_enabled.opensearch_service_domains_access_control_enabled.opensearch_client",
-            new=opensearch_client,
+            new=OpenSearchService(mocked_aws_provider),
         ):
             from prowler.providers.aws.services.opensearch.opensearch_service_domains_access_control_enabled.opensearch_service_domains_access_control_enabled import (
                 opensearch_service_domains_access_control_enabled,
@@ -30,22 +36,26 @@ class Test_opensearch_service_domains_access_control_enabled:
             result = check.execute()
             assert len(result) == 0
 
+    @mock_aws
     def test_no_fine_grained_access_enabled(self):
-        opensearch_client = mock.MagicMock
-        opensearch_client.opensearch_domains = []
-        opensearch_client.opensearch_domains.append(
-            OpenSearchDomain(
-                name=domain_name, region=AWS_REGION_EU_WEST_1, arn=domain_arn
-            )
+        opensearch_client = client("opensearch", region_name=AWS_REGION_US_EAST_1)
+        domain = opensearch_client.create_domain(
+            DomainName="test-domain",
+            AdvancedSecurityOptions={"Enabled": False},
         )
-        opensearch_client.opensearch_domains[0].advanced_settings_enabled = False
+
+        from prowler.providers.aws.services.opensearch.opensearch_service import (
+            OpenSearchService,
+        )
+
+        mocked_aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
 
         with mock.patch(
-            "prowler.providers.aws.services.opensearch.opensearch_service.OpenSearchService",
-            new=opensearch_client,
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=mocked_aws_provider,
         ), mock.patch(
             "prowler.providers.aws.services.opensearch.opensearch_service_domains_access_control_enabled.opensearch_service_domains_access_control_enabled.opensearch_client",
-            new=opensearch_client,
+            new=OpenSearchService(mocked_aws_provider),
         ):
             from prowler.providers.aws.services.opensearch.opensearch_service_domains_access_control_enabled.opensearch_service_domains_access_control_enabled import (
                 opensearch_service_domains_access_control_enabled,
@@ -55,29 +65,36 @@ class Test_opensearch_service_domains_access_control_enabled:
             result = check.execute()
             assert len(result) == 1
             assert result[0].status == "FAIL"
-            assert search(
-                "does not have fine grained access control enabled.",
-                result[0].status_extended,
+            assert (
+                f"Opensearch domain {domain['DomainStatus']['DomainName']} does not have fine grained access control enabled."
+                == result[0].status_extended
+            )
+            assert result[0].resource_id == domain["DomainStatus"]["DomainName"]
+            assert (
+                result[0].resource_arn
+                == f"arn:aws:es:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:domain/{domain['DomainStatus']['DomainName']}"
             )
-            assert result[0].resource_id == domain_name
-            assert result[0].resource_arn == domain_arn
 
-    def test_logging_AUDIT_LOGS_enabled(self):
-        opensearch_client = mock.MagicMock
-        opensearch_client.opensearch_domains = []
-        opensearch_client.opensearch_domains.append(
-            OpenSearchDomain(
-                name=domain_name, region=AWS_REGION_EU_WEST_1, arn=domain_arn
-            )
+    @mock_aws
+    def test_fine_grained_access_enabled(self):
+        opensearch_client = client("opensearch", region_name=AWS_REGION_US_EAST_1)
+        domain = opensearch_client.create_domain(
+            DomainName="test-domain",
+            AdvancedSecurityOptions={"Enabled": True},
         )
-        opensearch_client.opensearch_domains[0].advanced_settings_enabled = True
+
+        from prowler.providers.aws.services.opensearch.opensearch_service import (
+            OpenSearchService,
+        )
+
+        mocked_aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
 
         with mock.patch(
-            "prowler.providers.aws.services.opensearch.opensearch_service.OpenSearchService",
-            new=opensearch_client,
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=mocked_aws_provider,
         ), mock.patch(
             "prowler.providers.aws.services.opensearch.opensearch_service_domains_access_control_enabled.opensearch_service_domains_access_control_enabled.opensearch_client",
-            new=opensearch_client,
+            new=OpenSearchService(mocked_aws_provider),
         ):
             from prowler.providers.aws.services.opensearch.opensearch_service_domains_access_control_enabled.opensearch_service_domains_access_control_enabled import (
                 opensearch_service_domains_access_control_enabled,
@@ -87,8 +104,12 @@ class Test_opensearch_service_domains_access_control_enabled:
             result = check.execute()
             assert len(result) == 1
             assert result[0].status == "PASS"
-            assert search(
-                "has fine grained access control enabled.", result[0].status_extended
+            assert (
+                f"Opensearch domain {domain['DomainStatus']['DomainName']} has fine grained access control enabled."
+                == result[0].status_extended
+            )
+            assert result[0].resource_id == domain["DomainStatus"]["DomainName"]
+            assert (
+                result[0].resource_arn
+                == f"arn:aws:es:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:domain/{domain['DomainStatus']['DomainName']}"
             )
-            assert result[0].resource_id == domain_name
-            assert result[0].resource_arn == domain_arn

tests/providers/aws/services/opensearch/opensearch_service_domains_audit_logging_enabled/opensearch_service_domains_audit_logging_enabled_test.py

@@ -1,27 +1,32 @@
-from re import search
 from unittest import mock
 
-from prowler.providers.aws.services.opensearch.opensearch_service import (
-    OpenSearchDomain,
-    PublishingLoggingOption,
-)
-from tests.providers.aws.utils import AWS_ACCOUNT_NUMBER, AWS_REGION_EU_WEST_1
+from boto3 import client
+from moto import mock_aws
 
-domain_name = "test-domain"
-domain_arn = f"arn:aws:es:us-west-2:{AWS_ACCOUNT_NUMBER}:domain/{domain_name}"
+from tests.providers.aws.utils import (
+    AWS_ACCOUNT_NUMBER,
+    AWS_REGION_US_EAST_1,
+    set_mocked_aws_provider,
+)
 
 
 class Test_opensearch_service_domains_audit_logging_enabled:
+    @mock_aws
     def test_no_domains(self):
-        opensearch_client = mock.MagicMock
-        opensearch_client.opensearch_domains = []
+        client("opensearch", region_name=AWS_REGION_US_EAST_1)
+
+        from prowler.providers.aws.services.opensearch.opensearch_service import (
+            OpenSearchService,
+        )
+
+        mocked_aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
 
         with mock.patch(
-            "prowler.providers.aws.services.opensearch.opensearch_service.OpenSearchService",
-            new=opensearch_client,
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=mocked_aws_provider,
         ), mock.patch(
             "prowler.providers.aws.services.opensearch.opensearch_service_domains_audit_logging_enabled.opensearch_service_domains_audit_logging_enabled.opensearch_client",
-            new=opensearch_client,
+            new=OpenSearchService(mocked_aws_provider),
         ):
             from prowler.providers.aws.services.opensearch.opensearch_service_domains_audit_logging_enabled.opensearch_service_domains_audit_logging_enabled import (
                 opensearch_service_domains_audit_logging_enabled,
@@ -31,22 +36,29 @@ class Test_opensearch_service_domains_audit_logging_enabled:
             result = check.execute()
             assert len(result) == 0
 
+    @mock_aws
     def test_no_logging_enabled(self):
-        opensearch_client = mock.MagicMock
-        opensearch_client.opensearch_domains = []
-        opensearch_client.opensearch_domains.append(
-            OpenSearchDomain(
-                name=domain_name, region=AWS_REGION_EU_WEST_1, arn=domain_arn
-            )
+        opensearch_client = client("opensearch", region_name=AWS_REGION_US_EAST_1)
+        domain = opensearch_client.create_domain(
+            DomainName="test-domain-no-logging",
+            LogPublishingOptions={
+                "SEARCH_SLOW_LOGS": {
+                    "CloudWatchLogsLogGroupArn": "arn:aws:logs:us-east-1:123456789012:log-group:search-logs:*"
+                },
+            },
         )
-        opensearch_client.opensearch_domains[0].logging = []
+        from prowler.providers.aws.services.opensearch.opensearch_service import (
+            OpenSearchService,
+        )
+
+        mocked_aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
 
         with mock.patch(
-            "prowler.providers.aws.services.opensearch.opensearch_service.OpenSearchService",
-            new=opensearch_client,
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=mocked_aws_provider,
         ), mock.patch(
             "prowler.providers.aws.services.opensearch.opensearch_service_domains_audit_logging_enabled.opensearch_service_domains_audit_logging_enabled.opensearch_client",
-            new=opensearch_client,
+            new=OpenSearchService(mocked_aws_provider),
         ):
             from prowler.providers.aws.services.opensearch.opensearch_service_domains_audit_logging_enabled.opensearch_service_domains_audit_logging_enabled import (
                 opensearch_service_domains_audit_logging_enabled,
@@ -56,29 +68,40 @@ class Test_opensearch_service_domains_audit_logging_enabled:
             result = check.execute()
             assert len(result) == 1
             assert result[0].status == "FAIL"
-            assert search("AUDIT_LOGS disabled", result[0].status_extended)
-            assert result[0].resource_id == domain_name
-            assert result[0].resource_arn == domain_arn
-
-    def test_logging_AUDIT_LOGS_enabled(self):
-        opensearch_client = mock.MagicMock
-        opensearch_client.opensearch_domains = []
-        opensearch_client.opensearch_domains.append(
-            OpenSearchDomain(
-                name=domain_name, region=AWS_REGION_EU_WEST_1, arn=domain_arn
+            assert (
+                result[0].status_extended
+                == "Opensearch domain test-domain-no-logging AUDIT_LOGS disabled."
             )
+            assert result[0].resource_id == domain["DomainStatus"]["DomainName"]
+            assert (
+                result[0].resource_arn
+                == f"arn:aws:es:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:domain/{domain['DomainStatus']['DomainName']}"
+            )
+
+    @mock_aws
+    def test_logging_AUDIT_LOGS_enabled(self):
+        opensearch_client = client("opensearch", region_name=AWS_REGION_US_EAST_1)
+        domain = opensearch_client.create_domain(
+            DomainName="test-domain-logging",
+            LogPublishingOptions={
+                "AUDIT_LOGS": {
+                    "CloudWatchLogsLogGroupArn": "arn:aws:logs:us-east-1:123456789012:log-group:audit-logs:*",
+                    "Enabled": True,
+                },
+            },
         )
-        opensearch_client.opensearch_domains[0].logging = []
-        opensearch_client.opensearch_domains[0].logging.append(
-            PublishingLoggingOption(name="AUDIT_LOGS", enabled=True)
+        from prowler.providers.aws.services.opensearch.opensearch_service import (
+            OpenSearchService,
         )
 
+        mocked_aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
+
         with mock.patch(
-            "prowler.providers.aws.services.opensearch.opensearch_service.OpenSearchService",
-            new=opensearch_client,
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=mocked_aws_provider,
         ), mock.patch(
             "prowler.providers.aws.services.opensearch.opensearch_service_domains_audit_logging_enabled.opensearch_service_domains_audit_logging_enabled.opensearch_client",
-            new=opensearch_client,
+            new=OpenSearchService(mocked_aws_provider),
         ):
             from prowler.providers.aws.services.opensearch.opensearch_service_domains_audit_logging_enabled.opensearch_service_domains_audit_logging_enabled import (
                 opensearch_service_domains_audit_logging_enabled,
@@ -88,6 +111,12 @@ class Test_opensearch_service_domains_audit_logging_enabled:
             result = check.execute()
             assert len(result) == 1
             assert result[0].status == "PASS"
-            assert search("AUDIT_LOGS enabled", result[0].status_extended)
-            assert result[0].resource_id == domain_name
-            assert result[0].resource_arn == domain_arn
+            assert (
+                result[0].status_extended
+                == "Opensearch domain test-domain-logging AUDIT_LOGS enabled."
+            )
+            assert result[0].resource_id == domain["DomainStatus"]["DomainName"]
+            assert (
+                result[0].resource_arn
+                == f"arn:aws:es:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:domain/{domain['DomainStatus']['DomainName']}"
+            )

tests/providers/aws/services/opensearch/opensearch_service_domains_cloudwatch_logging_enabled/opensearch_service_domains_cloudwatch_logging_enabled_test.py

@@ -1,27 +1,32 @@
-from re import search
 from unittest import mock
 
-from prowler.providers.aws.services.opensearch.opensearch_service import (
-    OpenSearchDomain,
-    PublishingLoggingOption,
-)
-from tests.providers.aws.utils import AWS_ACCOUNT_NUMBER, AWS_REGION_EU_WEST_1
+from boto3 import client
+from moto import mock_aws
 
-domain_name = "test-domain"
-domain_arn = f"arn:aws:es:us-west-2:{AWS_ACCOUNT_NUMBER}:domain/{domain_name}"
+from tests.providers.aws.utils import (
+    AWS_ACCOUNT_NUMBER,
+    AWS_REGION_US_EAST_1,
+    set_mocked_aws_provider,
+)
 
 
 class Test_opensearch_service_domains_cloudwatch_logging_enabled:
+    @mock_aws
     def test_no_domains(self):
-        opensearch_client = mock.MagicMock
-        opensearch_client.opensearch_domains = []
+        client("opensearch", region_name=AWS_REGION_US_EAST_1)
+
+        from prowler.providers.aws.services.opensearch.opensearch_service import (
+            OpenSearchService,
+        )
+
+        mocked_aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
 
         with mock.patch(
-            "prowler.providers.aws.services.opensearch.opensearch_service.OpenSearchService",
-            new=opensearch_client,
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=mocked_aws_provider,
         ), mock.patch(
             "prowler.providers.aws.services.opensearch.opensearch_service_domains_cloudwatch_logging_enabled.opensearch_service_domains_cloudwatch_logging_enabled.opensearch_client",
-            new=opensearch_client,
+            new=OpenSearchService(mocked_aws_provider),
         ):
             from prowler.providers.aws.services.opensearch.opensearch_service_domains_cloudwatch_logging_enabled.opensearch_service_domains_cloudwatch_logging_enabled import (
                 opensearch_service_domains_cloudwatch_logging_enabled,
@@ -31,22 +36,30 @@ class Test_opensearch_service_domains_cloudwatch_logging_enabled:
             result = check.execute()
             assert len(result) == 0
 
+    @mock_aws
     def test_no_logging_enabled(self):
-        opensearch_client = mock.MagicMock
-        opensearch_client.opensearch_domains = []
-        opensearch_client.opensearch_domains.append(
-            OpenSearchDomain(
-                name=domain_name, region=AWS_REGION_EU_WEST_1, arn=domain_arn
-            )
+        opensearch_client = client("opensearch", region_name=AWS_REGION_US_EAST_1)
+        domain = opensearch_client.create_domain(
+            DomainName="test-domain-no-logging",
+            LogPublishingOptions={
+                "AUDIT_LOGS": {
+                    "CloudWatchLogsLogGroupArn": "arn:aws:logs:us-east-1:123456789012:log-group:search-logs:*",
+                    "Enabled": True,
+                },
+            },
         )
-        opensearch_client.opensearch_domains[0].logging = []
+        from prowler.providers.aws.services.opensearch.opensearch_service import (
+            OpenSearchService,
+        )
+
+        mocked_aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
 
         with mock.patch(
-            "prowler.providers.aws.services.opensearch.opensearch_service.OpenSearchService",
-            new=opensearch_client,
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=mocked_aws_provider,
         ), mock.patch(
             "prowler.providers.aws.services.opensearch.opensearch_service_domains_cloudwatch_logging_enabled.opensearch_service_domains_cloudwatch_logging_enabled.opensearch_client",
-            new=opensearch_client,
+            new=OpenSearchService(mocked_aws_provider),
         ):
             from prowler.providers.aws.services.opensearch.opensearch_service_domains_cloudwatch_logging_enabled.opensearch_service_domains_cloudwatch_logging_enabled import (
                 opensearch_service_domains_cloudwatch_logging_enabled,
@@ -56,32 +69,40 @@ class Test_opensearch_service_domains_cloudwatch_logging_enabled:
             result = check.execute()
             assert len(result) == 1
             assert result[0].status == "FAIL"
-            assert search(
-                "SEARCH_SLOW_LOGS and INDEX_SLOW_LOGS disabled",
-                result[0].status_extended,
+            assert (
+                result[0].status_extended
+                == "Opensearch domain test-domain-no-logging SEARCH_SLOW_LOGS and INDEX_SLOW_LOGS disabled."
+            )
+            assert result[0].resource_id == domain["DomainStatus"]["DomainName"]
+            assert (
+                result[0].resource_arn
+                == f"arn:aws:es:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:domain/{domain['DomainStatus']['DomainName']}"
             )
-            assert result[0].resource_id == domain_name
-            assert result[0].resource_arn == domain_arn
 
+    @mock_aws
     def test_logging_SEARCH_SLOW_LOGS_enabled(self):
-        opensearch_client = mock.MagicMock
-        opensearch_client.opensearch_domains = []
-        opensearch_client.opensearch_domains.append(
-            OpenSearchDomain(
-                name=domain_name, region=AWS_REGION_EU_WEST_1, arn=domain_arn
-            )
+        opensearch_client = client("opensearch", region_name=AWS_REGION_US_EAST_1)
+        domain = opensearch_client.create_domain(
+            DomainName="test-domain-no-index-logging",
+            LogPublishingOptions={
+                "SEARCH_SLOW_LOGS": {
+                    "CloudWatchLogsLogGroupArn": "arn:aws:logs:us-east-1:123456789012:log-group:search-logs:*",
+                    "Enabled": True,
+                },
+            },
         )
-        opensearch_client.opensearch_domains[0].logging = []
-        opensearch_client.opensearch_domains[0].logging.append(
-            PublishingLoggingOption(name="SEARCH_SLOW_LOGS", enabled=True)
+        from prowler.providers.aws.services.opensearch.opensearch_service import (
+            OpenSearchService,
         )
 
+        mocked_aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
+
         with mock.patch(
-            "prowler.providers.aws.services.opensearch.opensearch_service.OpenSearchService",
-            new=opensearch_client,
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=mocked_aws_provider,
         ), mock.patch(
             "prowler.providers.aws.services.opensearch.opensearch_service_domains_cloudwatch_logging_enabled.opensearch_service_domains_cloudwatch_logging_enabled.opensearch_client",
-            new=opensearch_client,
+            new=OpenSearchService(mocked_aws_provider),
         ):
             from prowler.providers.aws.services.opensearch.opensearch_service_domains_cloudwatch_logging_enabled.opensearch_service_domains_cloudwatch_logging_enabled import (
                 opensearch_service_domains_cloudwatch_logging_enabled,
@@ -91,32 +112,40 @@ class Test_opensearch_service_domains_cloudwatch_logging_enabled:
             result = check.execute()
             assert len(result) == 1
             assert result[0].status == "FAIL"
-            assert search(
-                "SEARCH_SLOW_LOGS enabled but INDEX_SLOW_LOGS disabled",
-                result[0].status_extended,
+            assert (
+                result[0].status_extended
+                == "Opensearch domain test-domain-no-index-logging SEARCH_SLOW_LOGS enabled but INDEX_SLOW_LOGS disabled."
+            )
+            assert result[0].resource_id == domain["DomainStatus"]["DomainName"]
+            assert (
+                result[0].resource_arn
+                == f"arn:aws:es:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:domain/{domain['DomainStatus']['DomainName']}"
             )
-            assert result[0].resource_id == domain_name
-            assert result[0].resource_arn == domain_arn
 
+    @mock_aws
     def test_logging_INDEX_SLOW_LOGS_enabled(self):
-        opensearch_client = mock.MagicMock
-        opensearch_client.opensearch_domains = []
-        opensearch_client.opensearch_domains.append(
-            OpenSearchDomain(
-                name=domain_name, region=AWS_REGION_EU_WEST_1, arn=domain_arn
-            )
+        opensearch_client = client("opensearch", region_name=AWS_REGION_US_EAST_1)
+        domain = opensearch_client.create_domain(
+            DomainName="test-domain-no-search-logging",
+            LogPublishingOptions={
+                "INDEX_SLOW_LOGS": {
+                    "CloudWatchLogsLogGroupArn": "arn:aws:logs:us-east-1:123456789012:log-group:search-logs:*",
+                    "Enabled": True,
+                },
+            },
         )
-        opensearch_client.opensearch_domains[0].logging = []
-        opensearch_client.opensearch_domains[0].logging.append(
-            PublishingLoggingOption(name="INDEX_SLOW_LOGS", enabled=True)
+        from prowler.providers.aws.services.opensearch.opensearch_service import (
+            OpenSearchService,
         )
 
+        mocked_aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
+
         with mock.patch(
-            "prowler.providers.aws.services.opensearch.opensearch_service.OpenSearchService",
-            new=opensearch_client,
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=mocked_aws_provider,
         ), mock.patch(
             "prowler.providers.aws.services.opensearch.opensearch_service_domains_cloudwatch_logging_enabled.opensearch_service_domains_cloudwatch_logging_enabled.opensearch_client",
-            new=opensearch_client,
+            new=OpenSearchService(mocked_aws_provider),
         ):
             from prowler.providers.aws.services.opensearch.opensearch_service_domains_cloudwatch_logging_enabled.opensearch_service_domains_cloudwatch_logging_enabled import (
                 opensearch_service_domains_cloudwatch_logging_enabled,
@@ -126,34 +155,44 @@ class Test_opensearch_service_domains_cloudwatch_logging_enabled:
             result = check.execute()
             assert len(result) == 1
             assert result[0].status == "FAIL"
-            assert search(
-                "INDEX_SLOW_LOGS enabled but SEARCH_SLOW_LOGS disabled",
-                result[0].status_extended,
+            assert (
+                result[0].status_extended
+                == "Opensearch domain test-domain-no-search-logging INDEX_SLOW_LOGS enabled but SEARCH_SLOW_LOGS disabled."
+            )
+            assert result[0].resource_id == domain["DomainStatus"]["DomainName"]
+            assert (
+                result[0].resource_arn
+                == f"arn:aws:es:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:domain/{domain['DomainStatus']['DomainName']}"
             )
-            assert result[0].resource_id == domain_name
-            assert result[0].resource_arn == domain_arn
 
+    @mock_aws
     def test_logging_INDEX_SLOW_LOGS_and_SEARCH_SLOW_LOGS_enabled(self):
-        opensearch_client = mock.MagicMock
-        opensearch_client.opensearch_domains = []
-        opensearch_client.opensearch_domains.append(
-            OpenSearchDomain(
-                name=domain_name, region=AWS_REGION_EU_WEST_1, arn=domain_arn
-            )
+        opensearch_client = client("opensearch", region_name=AWS_REGION_US_EAST_1)
+        domain = opensearch_client.create_domain(
+            DomainName="test-domain-logging",
+            LogPublishingOptions={
+                "SEARCH_SLOW_LOGS": {
+                    "CloudWatchLogsLogGroupArn": "arn:aws:logs:us-east-1:123456789012:log-group:search-logs:*",
+                    "Enabled": True,
+                },
+                "INDEX_SLOW_LOGS": {
+                    "CloudWatchLogsLogGroupArn": "arn:aws:logs:us-east-1:123456789012:log-group:search-logs:*",
+                    "Enabled": True,
+                },
+            },
         )
-        opensearch_client.opensearch_domains[0].logging = []
-        logging_options = [
-            PublishingLoggingOption(name="INDEX_SLOW_LOGS", enabled=True),
-            PublishingLoggingOption(name="SEARCH_SLOW_LOGS", enabled=True),
-        ]
-        opensearch_client.opensearch_domains[0].logging.extend(logging_options)
+        from prowler.providers.aws.services.opensearch.opensearch_service import (
+            OpenSearchService,
+        )
+
+        mocked_aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
 
         with mock.patch(
-            "prowler.providers.aws.services.opensearch.opensearch_service.OpenSearchService",
-            new=opensearch_client,
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=mocked_aws_provider,
         ), mock.patch(
             "prowler.providers.aws.services.opensearch.opensearch_service_domains_cloudwatch_logging_enabled.opensearch_service_domains_cloudwatch_logging_enabled.opensearch_client",
-            new=opensearch_client,
+            new=OpenSearchService(mocked_aws_provider),
         ):
             from prowler.providers.aws.services.opensearch.opensearch_service_domains_cloudwatch_logging_enabled.opensearch_service_domains_cloudwatch_logging_enabled import (
                 opensearch_service_domains_cloudwatch_logging_enabled,
@@ -163,9 +202,12 @@ class Test_opensearch_service_domains_cloudwatch_logging_enabled:
             result = check.execute()
             assert len(result) == 1
             assert result[0].status == "PASS"
-            assert search(
-                "SEARCH_SLOW_LOGS and INDEX_SLOW_LOGS enabled",
-                result[0].status_extended,
+            assert (
+                result[0].status_extended
+                == "Opensearch domain test-domain-logging SEARCH_SLOW_LOGS and INDEX_SLOW_LOGS enabled."
+            )
+            assert result[0].resource_id == domain["DomainStatus"]["DomainName"]
+            assert (
+                result[0].resource_arn
+                == f"arn:aws:es:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:domain/{domain['DomainStatus']['DomainName']}"
             )
-            assert result[0].resource_id == domain_name
-            assert result[0].resource_arn == domain_arn

tests/providers/aws/services/opensearch/opensearch_service_domains_encryption_at_rest_enabled/opensearch_service_domains_encryption_at_rest_enabled_test.py

@@ -1,26 +1,32 @@
-from re import search
 from unittest import mock
 
-from prowler.providers.aws.services.opensearch.opensearch_service import (
-    OpenSearchDomain,
-)
-from tests.providers.aws.utils import AWS_ACCOUNT_NUMBER, AWS_REGION_EU_WEST_1
+from boto3 import client
+from moto import mock_aws
 
-domain_name = "test-domain"
-domain_arn = f"arn:aws:es:us-west-2:{AWS_ACCOUNT_NUMBER}:domain/{domain_name}"
+from tests.providers.aws.utils import (
+    AWS_ACCOUNT_NUMBER,
+    AWS_REGION_US_EAST_1,
+    set_mocked_aws_provider,
+)
 
 
 class Test_opensearch_service_domains_encryption_at_rest_enabled:
+    @mock_aws
     def test_no_domains(self):
-        opensearch_client = mock.MagicMock
-        opensearch_client.opensearch_domains = []
+        client("opensearch", region_name=AWS_REGION_US_EAST_1)
+
+        from prowler.providers.aws.services.opensearch.opensearch_service import (
+            OpenSearchService,
+        )
+
+        mocked_aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
 
         with mock.patch(
-            "prowler.providers.aws.services.opensearch.opensearch_service.OpenSearchService",
-            new=opensearch_client,
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=mocked_aws_provider,
         ), mock.patch(
             "prowler.providers.aws.services.opensearch.opensearch_service_domains_encryption_at_rest_enabled.opensearch_service_domains_encryption_at_rest_enabled.opensearch_client",
-            new=opensearch_client,
+            new=OpenSearchService(mocked_aws_provider),
         ):
             from prowler.providers.aws.services.opensearch.opensearch_service_domains_encryption_at_rest_enabled.opensearch_service_domains_encryption_at_rest_enabled import (
                 opensearch_service_domains_encryption_at_rest_enabled,
@@ -30,25 +36,25 @@ class Test_opensearch_service_domains_encryption_at_rest_enabled:
             result = check.execute()
             assert len(result) == 0
 
+    @mock_aws
     def test_no_encryption_at_rest_enabled(self):
-        opensearch_client = mock.MagicMock
-        opensearch_client.opensearch_domains = []
-        opensearch_client.opensearch_domains.append(
-            OpenSearchDomain(
-                name=domain_name,
-                region=AWS_REGION_EU_WEST_1,
-                arn=domain_arn,
-                encryption_at_rest=False,
-            )
+        opensearch_client = client("opensearch", region_name=AWS_REGION_US_EAST_1)
+        domain = opensearch_client.create_domain(
+            DomainName="test-domain-no-encryption",
+            EncryptionAtRestOptions={"Enabled": False},
         )
-        opensearch_client.opensearch_domains[0].logging = []
+        from prowler.providers.aws.services.opensearch.opensearch_service import (
+            OpenSearchService,
+        )
+
+        mocked_aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
 
         with mock.patch(
-            "prowler.providers.aws.services.opensearch.opensearch_service.OpenSearchService",
-            new=opensearch_client,
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=mocked_aws_provider,
         ), mock.patch(
             "prowler.providers.aws.services.opensearch.opensearch_service_domains_encryption_at_rest_enabled.opensearch_service_domains_encryption_at_rest_enabled.opensearch_client",
-            new=opensearch_client,
+            new=OpenSearchService(mocked_aws_provider),
         ):
             from prowler.providers.aws.services.opensearch.opensearch_service_domains_encryption_at_rest_enabled.opensearch_service_domains_encryption_at_rest_enabled import (
                 opensearch_service_domains_encryption_at_rest_enabled,
@@ -58,31 +64,35 @@ class Test_opensearch_service_domains_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].status_extended
+                == "Opensearch domain test-domain-no-encryption does not have encryption at-rest enabled."
+            )
+            assert result[0].resource_id == domain["DomainStatus"]["DomainName"]
+            assert (
+                result[0].resource_arn
+                == f"arn:aws:es:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:domain/{domain['DomainStatus']['DomainName']}"
             )
-            assert result[0].resource_id == domain_name
-            assert result[0].resource_arn == domain_arn
 
+    @mock_aws
     def test_encryption_at_rest_enabled(self):
-        opensearch_client = mock.MagicMock
-        opensearch_client.opensearch_domains = []
-        opensearch_client.opensearch_domains.append(
-            OpenSearchDomain(
-                name=domain_name,
-                region=AWS_REGION_EU_WEST_1,
-                arn=domain_arn,
-                encryption_at_rest=True,
-            )
+        opensearch_client = client("opensearch", region_name=AWS_REGION_US_EAST_1)
+        domain = opensearch_client.create_domain(
+            DomainName="test-domain-encryption-at-rest",
+            EncryptionAtRestOptions={"Enabled": True},
         )
-        opensearch_client.opensearch_domains[0].logging = []
+        from prowler.providers.aws.services.opensearch.opensearch_service import (
+            OpenSearchService,
+        )
+
+        mocked_aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
 
         with mock.patch(
-            "prowler.providers.aws.services.opensearch.opensearch_service.OpenSearchService",
-            new=opensearch_client,
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=mocked_aws_provider,
         ), mock.patch(
             "prowler.providers.aws.services.opensearch.opensearch_service_domains_encryption_at_rest_enabled.opensearch_service_domains_encryption_at_rest_enabled.opensearch_client",
-            new=opensearch_client,
+            new=OpenSearchService(mocked_aws_provider),
         ):
             from prowler.providers.aws.services.opensearch.opensearch_service_domains_encryption_at_rest_enabled.opensearch_service_domains_encryption_at_rest_enabled import (
                 opensearch_service_domains_encryption_at_rest_enabled,
@@ -92,6 +102,12 @@ class Test_opensearch_service_domains_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 == domain_name
-            assert result[0].resource_arn == domain_arn
+            assert (
+                result[0].status_extended
+                == "Opensearch domain test-domain-encryption-at-rest has encryption at-rest enabled."
+            )
+            assert result[0].resource_id == domain["DomainStatus"]["DomainName"]
+            assert (
+                result[0].resource_arn
+                == f"arn:aws:es:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:domain/{domain['DomainStatus']['DomainName']}"
+            )

tests/providers/aws/services/opensearch/opensearch_service_domains_fault_tolerant_data_nodes/opensearch_service_domains_fault_tolerant_data_nodes_data_nodes_test.py

@@ -0,0 +1,193 @@
+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,
+)
+
+
+class Test_opensearch_service_domains_fault_tolerant_data_nodes:
+    @mock_aws
+    def test_no_domains(self):
+        client("opensearch", region_name=AWS_REGION_US_EAST_1)
+
+        from prowler.providers.aws.services.opensearch.opensearch_service import (
+            OpenSearchService,
+        )
+
+        mocked_aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=mocked_aws_provider,
+        ), mock.patch(
+            "prowler.providers.aws.services.opensearch.opensearch_service_domains_fault_tolerant_data_nodes.opensearch_service_domains_fault_tolerant_data_nodes.opensearch_client",
+            new=OpenSearchService(mocked_aws_provider),
+        ):
+            from prowler.providers.aws.services.opensearch.opensearch_service_domains_fault_tolerant_data_nodes.opensearch_service_domains_fault_tolerant_data_nodes import (
+                opensearch_service_domains_fault_tolerant_data_nodes,
+            )
+
+            check = opensearch_service_domains_fault_tolerant_data_nodes()
+            result = check.execute()
+            assert len(result) == 0
+
+    @mock_aws
+    def test_instance_count_less_than_three_zoneawareness_enabled(self):
+        opensearch_client = client("opensearch", region_name=AWS_REGION_US_EAST_1)
+        domain = opensearch_client.create_domain(
+            DomainName="test-domain",
+            ClusterConfig={"ZoneAwarenessEnabled": True, "InstanceCount": 2},
+        )
+
+        from prowler.providers.aws.services.opensearch.opensearch_service import (
+            OpenSearchService,
+        )
+
+        mocked_aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=mocked_aws_provider,
+        ), mock.patch(
+            "prowler.providers.aws.services.opensearch.opensearch_service_domains_fault_tolerant_data_nodes.opensearch_service_domains_fault_tolerant_data_nodes.opensearch_client",
+            new=OpenSearchService(mocked_aws_provider),
+        ):
+            from prowler.providers.aws.services.opensearch.opensearch_service_domains_fault_tolerant_data_nodes.opensearch_service_domains_fault_tolerant_data_nodes import (
+                opensearch_service_domains_fault_tolerant_data_nodes,
+            )
+
+            check = opensearch_service_domains_fault_tolerant_data_nodes()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert (
+                result[0].status_extended
+                == "Opensearch domain test-domain is not fault tolerant as it has cross-zone replication (Zone Awareness) enabled, but only 2 data nodes."
+            )
+            assert result[0].resource_id == domain["DomainStatus"]["DomainName"]
+            assert (
+                result[0].resource_arn
+                == f"arn:aws:es:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:domain/{domain['DomainStatus']['DomainName']}"
+            )
+
+    @mock_aws
+    def test_instance_count_more_than_three_zoneawareness_not_enabled(self):
+        opensearch_client = client("opensearch", region_name=AWS_REGION_US_EAST_1)
+        domain = opensearch_client.create_domain(
+            DomainName="test-domain",
+            ClusterConfig={"ZoneAwarenessEnabled": False, "InstanceCount": 3},
+        )
+
+        from prowler.providers.aws.services.opensearch.opensearch_service import (
+            OpenSearchService,
+        )
+
+        mocked_aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=mocked_aws_provider,
+        ), mock.patch(
+            "prowler.providers.aws.services.opensearch.opensearch_service_domains_fault_tolerant_data_nodes.opensearch_service_domains_fault_tolerant_data_nodes.opensearch_client",
+            new=OpenSearchService(mocked_aws_provider),
+        ):
+            from prowler.providers.aws.services.opensearch.opensearch_service_domains_fault_tolerant_data_nodes.opensearch_service_domains_fault_tolerant_data_nodes import (
+                opensearch_service_domains_fault_tolerant_data_nodes,
+            )
+
+            check = opensearch_service_domains_fault_tolerant_data_nodes()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert (
+                result[0].status_extended
+                == "Opensearch domain test-domain is not fault tolerant as it has 3 data nodes, but cross-zone replication (Zone Awareness) is not enabled."
+            )
+            assert result[0].resource_id == domain["DomainStatus"]["DomainName"]
+            assert (
+                result[0].resource_arn
+                == f"arn:aws:es:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:domain/{domain['DomainStatus']['DomainName']}"
+            )
+
+    @mock_aws
+    def test_instance_count_less_than_three_zoneawareness_not_enabled(self):
+        opensearch_client = client("opensearch", region_name=AWS_REGION_US_EAST_1)
+        domain = opensearch_client.create_domain(
+            DomainName="test-domain",
+            ClusterConfig={"ZoneAwarenessEnabled": False, "InstanceCount": 2},
+        )
+
+        from prowler.providers.aws.services.opensearch.opensearch_service import (
+            OpenSearchService,
+        )
+
+        mocked_aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=mocked_aws_provider,
+        ), mock.patch(
+            "prowler.providers.aws.services.opensearch.opensearch_service_domains_fault_tolerant_data_nodes.opensearch_service_domains_fault_tolerant_data_nodes.opensearch_client",
+            new=OpenSearchService(mocked_aws_provider),
+        ):
+            from prowler.providers.aws.services.opensearch.opensearch_service_domains_fault_tolerant_data_nodes.opensearch_service_domains_fault_tolerant_data_nodes import (
+                opensearch_service_domains_fault_tolerant_data_nodes,
+            )
+
+            check = opensearch_service_domains_fault_tolerant_data_nodes()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert (
+                result[0].status_extended
+                == "Opensearch domain test-domain is not fault tolerant as it has less than 3 data nodes and cross-zone replication (Zone Awareness) is not enabled."
+            )
+            assert result[0].resource_id == domain["DomainStatus"]["DomainName"]
+            assert (
+                result[0].resource_arn
+                == f"arn:aws:es:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:domain/{domain['DomainStatus']['DomainName']}"
+            )
+
+    @mock_aws
+    def test_logging_instance_count_more_than_three_zoneawareness_enabled(self):
+        opensearch_client = client("opensearch", region_name=AWS_REGION_US_EAST_1)
+        domain = opensearch_client.create_domain(
+            DomainName="test-domain",
+            ClusterConfig={"ZoneAwarenessEnabled": True, "InstanceCount": 3},
+        )
+
+        from prowler.providers.aws.services.opensearch.opensearch_service import (
+            OpenSearchService,
+        )
+
+        mocked_aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=mocked_aws_provider,
+        ), mock.patch(
+            "prowler.providers.aws.services.opensearch.opensearch_service_domains_fault_tolerant_data_nodes.opensearch_service_domains_fault_tolerant_data_nodes.opensearch_client",
+            new=OpenSearchService(mocked_aws_provider),
+        ):
+            from prowler.providers.aws.services.opensearch.opensearch_service_domains_fault_tolerant_data_nodes.opensearch_service_domains_fault_tolerant_data_nodes import (
+                opensearch_service_domains_fault_tolerant_data_nodes,
+            )
+
+            check = opensearch_service_domains_fault_tolerant_data_nodes()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "PASS"
+            assert (
+                result[0].status_extended
+                == "Opensearch domain test-domain is fault tolerant with 3 data nodes and cross-zone replication (Zone Awareness) enabled."
+            )
+            assert result[0].resource_id == domain["DomainStatus"]["DomainName"]
+            assert (
+                result[0].resource_arn
+                == f"arn:aws:es:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:domain/{domain['DomainStatus']['DomainName']}"
+            )

tests/providers/aws/services/opensearch/opensearch_service_domains_https_communications_enforced/opensearch_service_domains_https_communications_enforced_test.py

@@ -1,26 +1,32 @@
-from re import search
 from unittest import mock
 
-from prowler.providers.aws.services.opensearch.opensearch_service import (
-    OpenSearchDomain,
-)
-from tests.providers.aws.utils import AWS_ACCOUNT_NUMBER, AWS_REGION_EU_WEST_1
+from boto3 import client
+from moto import mock_aws
 
-domain_name = "test-domain"
-domain_arn = f"arn:aws:es:us-west-2:{AWS_ACCOUNT_NUMBER}:domain/{domain_name}"
+from tests.providers.aws.utils import (
+    AWS_ACCOUNT_NUMBER,
+    AWS_REGION_US_EAST_1,
+    set_mocked_aws_provider,
+)
 
 
 class Test_opensearch_service_domains_https_communications_enforced:
+    @mock_aws
     def test_no_domains(self):
-        opensearch_client = mock.MagicMock
-        opensearch_client.opensearch_domains = []
+        client("opensearch", region_name=AWS_REGION_US_EAST_1)
+
+        from prowler.providers.aws.services.opensearch.opensearch_service import (
+            OpenSearchService,
+        )
+
+        mocked_aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
 
         with mock.patch(
-            "prowler.providers.aws.services.opensearch.opensearch_service.OpenSearchService",
-            new=opensearch_client,
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=mocked_aws_provider,
         ), mock.patch(
             "prowler.providers.aws.services.opensearch.opensearch_service_domains_https_communications_enforced.opensearch_service_domains_https_communications_enforced.opensearch_client",
-            new=opensearch_client,
+            new=OpenSearchService(mocked_aws_provider),
         ):
             from prowler.providers.aws.services.opensearch.opensearch_service_domains_https_communications_enforced.opensearch_service_domains_https_communications_enforced import (
                 opensearch_service_domains_https_communications_enforced,
@@ -30,25 +36,27 @@ class Test_opensearch_service_domains_https_communications_enforced:
             result = check.execute()
             assert len(result) == 0
 
+    @mock_aws
     def test_no_https_enabled(self):
-        opensearch_client = mock.MagicMock
-        opensearch_client.opensearch_domains = []
-        opensearch_client.opensearch_domains.append(
-            OpenSearchDomain(
-                name=domain_name,
-                region=AWS_REGION_EU_WEST_1,
-                arn=domain_arn,
-                enforce_https=False,
-            )
+        opensearch_client = client("opensearch", region_name=AWS_REGION_US_EAST_1)
+        domain = opensearch_client.create_domain(
+            DomainName="test-domain-no-https",
+            DomainEndpointOptions={
+                "EnforceHTTPS": False,
+            },
         )
-        opensearch_client.opensearch_domains[0].logging = []
+        from prowler.providers.aws.services.opensearch.opensearch_service import (
+            OpenSearchService,
+        )
+
+        mocked_aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
 
         with mock.patch(
-            "prowler.providers.aws.services.opensearch.opensearch_service.OpenSearchService",
-            new=opensearch_client,
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=mocked_aws_provider,
         ), mock.patch(
             "prowler.providers.aws.services.opensearch.opensearch_service_domains_https_communications_enforced.opensearch_service_domains_https_communications_enforced.opensearch_client",
-            new=opensearch_client,
+            new=OpenSearchService(mocked_aws_provider),
         ):
             from prowler.providers.aws.services.opensearch.opensearch_service_domains_https_communications_enforced.opensearch_service_domains_https_communications_enforced import (
                 opensearch_service_domains_https_communications_enforced,
@@ -58,31 +66,37 @@ class Test_opensearch_service_domains_https_communications_enforced:
             result = check.execute()
             assert len(result) == 1
             assert result[0].status == "FAIL"
-            assert search(
-                "does not have enforce HTTPS enabled", result[0].status_extended
+            assert (
+                result[0].status_extended
+                == "Opensearch domain test-domain-no-https does not have enforce HTTPS enabled."
+            )
+            assert result[0].resource_id == domain["DomainStatus"]["DomainName"]
+            assert (
+                result[0].resource_arn
+                == f"arn:aws:es:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:domain/{domain['DomainStatus']['DomainName']}"
             )
-            assert result[0].resource_id == domain_name
-            assert result[0].resource_arn == domain_arn
 
+    @mock_aws
     def test_https_enabled(self):
-        opensearch_client = mock.MagicMock
-        opensearch_client.opensearch_domains = []
-        opensearch_client.opensearch_domains.append(
-            OpenSearchDomain(
-                name=domain_name,
-                region=AWS_REGION_EU_WEST_1,
-                arn=domain_arn,
-                enforce_https=True,
-            )
+        opensearch_client = client("opensearch", region_name=AWS_REGION_US_EAST_1)
+        domain = opensearch_client.create_domain(
+            DomainName="test-domain-https",
+            DomainEndpointOptions={
+                "EnforceHTTPS": True,
+            },
         )
-        opensearch_client.opensearch_domains[0].logging = []
+        from prowler.providers.aws.services.opensearch.opensearch_service import (
+            OpenSearchService,
+        )
+
+        mocked_aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
 
         with mock.patch(
-            "prowler.providers.aws.services.opensearch.opensearch_service.OpenSearchService",
-            new=opensearch_client,
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=mocked_aws_provider,
         ), mock.patch(
             "prowler.providers.aws.services.opensearch.opensearch_service_domains_https_communications_enforced.opensearch_service_domains_https_communications_enforced.opensearch_client",
-            new=opensearch_client,
+            new=OpenSearchService(mocked_aws_provider),
         ):
             from prowler.providers.aws.services.opensearch.opensearch_service_domains_https_communications_enforced.opensearch_service_domains_https_communications_enforced import (
                 opensearch_service_domains_https_communications_enforced,
@@ -92,6 +106,12 @@ class Test_opensearch_service_domains_https_communications_enforced:
             result = check.execute()
             assert len(result) == 1
             assert result[0].status == "PASS"
-            assert search("has enforce HTTPS enabled", result[0].status_extended)
-            assert result[0].resource_id == domain_name
-            assert result[0].resource_arn == domain_arn
+            assert (
+                result[0].status_extended
+                == "Opensearch domain test-domain-https has enforce HTTPS enabled."
+            )
+            assert result[0].resource_id == domain["DomainStatus"]["DomainName"]
+            assert (
+                result[0].resource_arn
+                == f"arn:aws:es:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:domain/{domain['DomainStatus']['DomainName']}"
+            )

tests/providers/aws/services/opensearch/opensearch_service_domains_internal_user_database_enabled/opensearch_service_domains_internal_user_database_enabled_test.py

@@ -1,26 +1,32 @@
-from re import search
 from unittest import mock
 
-from prowler.providers.aws.services.opensearch.opensearch_service import (
-    OpenSearchDomain,
-)
-from tests.providers.aws.utils import AWS_ACCOUNT_NUMBER, AWS_REGION_EU_WEST_1
+from boto3 import client
+from moto import mock_aws
 
-domain_name = "test-domain"
-domain_arn = f"arn:aws:es:us-west-2:{AWS_ACCOUNT_NUMBER}:domain/{domain_name}"
+from tests.providers.aws.utils import (
+    AWS_ACCOUNT_NUMBER,
+    AWS_REGION_US_EAST_1,
+    set_mocked_aws_provider,
+)
 
 
 class Test_opensearch_service_domains_internal_user_database_enabled:
+    @mock_aws
     def test_no_domains(self):
-        opensearch_client = mock.MagicMock
-        opensearch_client.opensearch_domains = []
+        client("opensearch", region_name=AWS_REGION_US_EAST_1)
+
+        from prowler.providers.aws.services.opensearch.opensearch_service import (
+            OpenSearchService,
+        )
+
+        mocked_aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
 
         with mock.patch(
-            "prowler.providers.aws.services.opensearch.opensearch_service.OpenSearchService",
-            new=opensearch_client,
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=mocked_aws_provider,
         ), mock.patch(
             "prowler.providers.aws.services.opensearch.opensearch_service_domains_internal_user_database_enabled.opensearch_service_domains_internal_user_database_enabled.opensearch_client",
-            new=opensearch_client,
+            new=OpenSearchService(mocked_aws_provider),
         ):
             from prowler.providers.aws.services.opensearch.opensearch_service_domains_internal_user_database_enabled.opensearch_service_domains_internal_user_database_enabled import (
                 opensearch_service_domains_internal_user_database_enabled,
@@ -30,25 +36,25 @@ class Test_opensearch_service_domains_internal_user_database_enabled:
             result = check.execute()
             assert len(result) == 0
 
+    @mock_aws
     def test_internal_database_disabled(self):
-        opensearch_client = mock.MagicMock
-        opensearch_client.opensearch_domains = []
-        opensearch_client.opensearch_domains.append(
-            OpenSearchDomain(
-                name=domain_name,
-                region=AWS_REGION_EU_WEST_1,
-                arn=domain_arn,
-                internal_user_database=False,
-            )
+        opensearch_client = client("opensearch", region_name=AWS_REGION_US_EAST_1)
+        domain = opensearch_client.create_domain(
+            DomainName="test-domain",
+            AdvancedSecurityOptions={"InternalUserDatabaseEnabled": False},
         )
-        opensearch_client.opensearch_domains[0].logging = []
+        from prowler.providers.aws.services.opensearch.opensearch_service import (
+            OpenSearchService,
+        )
+
+        mocked_aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
 
         with mock.patch(
-            "prowler.providers.aws.services.opensearch.opensearch_service.OpenSearchService",
-            new=opensearch_client,
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=mocked_aws_provider,
         ), mock.patch(
             "prowler.providers.aws.services.opensearch.opensearch_service_domains_internal_user_database_enabled.opensearch_service_domains_internal_user_database_enabled.opensearch_client",
-            new=opensearch_client,
+            new=OpenSearchService(mocked_aws_provider),
         ):
             from prowler.providers.aws.services.opensearch.opensearch_service_domains_internal_user_database_enabled.opensearch_service_domains_internal_user_database_enabled import (
                 opensearch_service_domains_internal_user_database_enabled,
@@ -58,32 +64,35 @@ class Test_opensearch_service_domains_internal_user_database_enabled:
             result = check.execute()
             assert len(result) == 1
             assert result[0].status == "PASS"
-            assert search(
-                "does not have internal user database enabled",
-                result[0].status_extended,
+            assert (
+                result[0].status_extended
+                == "Opensearch domain test-domain does not have internal user database enabled."
+            )
+            assert result[0].resource_id == domain["DomainStatus"]["DomainName"]
+            assert (
+                result[0].resource_arn
+                == f"arn:aws:es:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:domain/{domain['DomainStatus']['DomainName']}"
             )
-            assert result[0].resource_id == domain_name
-            assert result[0].resource_arn == domain_arn
 
+    @mock_aws
     def test_internal_database_enabled(self):
-        opensearch_client = mock.MagicMock
-        opensearch_client.opensearch_domains = []
-        opensearch_client.opensearch_domains.append(
-            OpenSearchDomain(
-                name=domain_name,
-                region=AWS_REGION_EU_WEST_1,
-                arn=domain_arn,
-                internal_user_database=True,
-            )
+        opensearch_client = client("opensearch", region_name=AWS_REGION_US_EAST_1)
+        domain = opensearch_client.create_domain(
+            DomainName="test-domain",
+            AdvancedSecurityOptions={"InternalUserDatabaseEnabled": True},
         )
-        opensearch_client.opensearch_domains[0].logging = []
+        from prowler.providers.aws.services.opensearch.opensearch_service import (
+            OpenSearchService,
+        )
+
+        mocked_aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
 
         with mock.patch(
-            "prowler.providers.aws.services.opensearch.opensearch_service.OpenSearchService",
-            new=opensearch_client,
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=mocked_aws_provider,
         ), mock.patch(
             "prowler.providers.aws.services.opensearch.opensearch_service_domains_internal_user_database_enabled.opensearch_service_domains_internal_user_database_enabled.opensearch_client",
-            new=opensearch_client,
+            new=OpenSearchService(mocked_aws_provider),
         ):
             from prowler.providers.aws.services.opensearch.opensearch_service_domains_internal_user_database_enabled.opensearch_service_domains_internal_user_database_enabled import (
                 opensearch_service_domains_internal_user_database_enabled,
@@ -93,8 +102,12 @@ class Test_opensearch_service_domains_internal_user_database_enabled:
             result = check.execute()
             assert len(result) == 1
             assert result[0].status == "FAIL"
-            assert search(
-                "has internal user database enabled", result[0].status_extended
+            assert (
+                result[0].status_extended
+                == "Opensearch domain test-domain has internal user database enabled."
+            )
+            assert result[0].resource_id == domain["DomainStatus"]["DomainName"]
+            assert (
+                result[0].resource_arn
+                == f"arn:aws:es:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:domain/{domain['DomainStatus']['DomainName']}"
             )
-            assert result[0].resource_id == domain_name
-            assert result[0].resource_arn == domain_arn

tests/providers/aws/services/opensearch/opensearch_service_domains_node_to_node_encryption_enabled/opensearch_service_domains_node_to_node_encryption_enabled_test.py

@@ -1,26 +1,32 @@
-from re import search
 from unittest import mock
 
-from prowler.providers.aws.services.opensearch.opensearch_service import (
-    OpenSearchDomain,
-)
-from tests.providers.aws.utils import AWS_ACCOUNT_NUMBER, AWS_REGION_EU_WEST_1
+from boto3 import client
+from moto import mock_aws
 
-domain_name = "test-domain"
-domain_arn = f"arn:aws:es:us-west-2:{AWS_ACCOUNT_NUMBER}:domain/{domain_name}"
+from tests.providers.aws.utils import (
+    AWS_ACCOUNT_NUMBER,
+    AWS_REGION_US_EAST_1,
+    set_mocked_aws_provider,
+)
 
 
 class Test_opensearch_service_domains_node_to_node_encryption_enabled:
+    @mock_aws
     def test_no_domains(self):
-        opensearch_client = mock.MagicMock
-        opensearch_client.opensearch_domains = []
+        client("opensearch", region_name=AWS_REGION_US_EAST_1)
+
+        from prowler.providers.aws.services.opensearch.opensearch_service import (
+            OpenSearchService,
+        )
+
+        mocked_aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
 
         with mock.patch(
-            "prowler.providers.aws.services.opensearch.opensearch_service.OpenSearchService",
-            new=opensearch_client,
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=mocked_aws_provider,
         ), mock.patch(
             "prowler.providers.aws.services.opensearch.opensearch_service_domains_node_to_node_encryption_enabled.opensearch_service_domains_node_to_node_encryption_enabled.opensearch_client",
-            new=opensearch_client,
+            new=OpenSearchService(mocked_aws_provider),
         ):
             from prowler.providers.aws.services.opensearch.opensearch_service_domains_node_to_node_encryption_enabled.opensearch_service_domains_node_to_node_encryption_enabled import (
                 opensearch_service_domains_node_to_node_encryption_enabled,
@@ -30,25 +36,25 @@ class Test_opensearch_service_domains_node_to_node_encryption_enabled:
             result = check.execute()
             assert len(result) == 0
 
+    @mock_aws
     def test_no_encryption_enabled(self):
-        opensearch_client = mock.MagicMock
-        opensearch_client.opensearch_domains = []
-        opensearch_client.opensearch_domains.append(
-            OpenSearchDomain(
-                name=domain_name,
-                region=AWS_REGION_EU_WEST_1,
-                arn=domain_arn,
-                node_to_node_encryption=False,
-            )
+        opensearch_client = client("opensearch", region_name=AWS_REGION_US_EAST_1)
+        domain = opensearch_client.create_domain(
+            DomainName="test-domain-no-encryption",
+            NodeToNodeEncryptionOptions={"Enabled": False},
         )
-        opensearch_client.opensearch_domains[0].logging = []
+        from prowler.providers.aws.services.opensearch.opensearch_service import (
+            OpenSearchService,
+        )
+
+        mocked_aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
 
         with mock.patch(
-            "prowler.providers.aws.services.opensearch.opensearch_service.OpenSearchService",
-            new=opensearch_client,
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=mocked_aws_provider,
         ), mock.patch(
             "prowler.providers.aws.services.opensearch.opensearch_service_domains_node_to_node_encryption_enabled.opensearch_service_domains_node_to_node_encryption_enabled.opensearch_client",
-            new=opensearch_client,
+            new=OpenSearchService(mocked_aws_provider),
         ):
             from prowler.providers.aws.services.opensearch.opensearch_service_domains_node_to_node_encryption_enabled.opensearch_service_domains_node_to_node_encryption_enabled import (
                 opensearch_service_domains_node_to_node_encryption_enabled,
@@ -58,32 +64,35 @@ class Test_opensearch_service_domains_node_to_node_encryption_enabled:
             result = check.execute()
             assert len(result) == 1
             assert result[0].status == "FAIL"
-            assert search(
-                "does not have node-to-node encryption enabled",
-                result[0].status_extended,
+            assert (
+                result[0].status_extended
+                == "Opensearch domain test-domain-no-encryption does not have node-to-node encryption enabled."
+            )
+            assert result[0].resource_id == domain["DomainStatus"]["DomainName"]
+            assert (
+                result[0].resource_arn
+                == f"arn:aws:es:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:domain/{domain['DomainStatus']['DomainName']}"
             )
-            assert result[0].resource_id == domain_name
-            assert result[0].resource_arn == domain_arn
 
+    @mock_aws
     def test_encryption_enabled(self):
-        opensearch_client = mock.MagicMock
-        opensearch_client.opensearch_domains = []
-        opensearch_client.opensearch_domains.append(
-            OpenSearchDomain(
-                name=domain_name,
-                region=AWS_REGION_EU_WEST_1,
-                arn=domain_arn,
-                node_to_node_encryption=True,
-            )
+        opensearch_client = client("opensearch", region_name=AWS_REGION_US_EAST_1)
+        domain = opensearch_client.create_domain(
+            DomainName="test-domain-encryption",
+            NodeToNodeEncryptionOptions={"Enabled": True},
         )
-        opensearch_client.opensearch_domains[0].logging = []
+        from prowler.providers.aws.services.opensearch.opensearch_service import (
+            OpenSearchService,
+        )
+
+        mocked_aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
 
         with mock.patch(
-            "prowler.providers.aws.services.opensearch.opensearch_service.OpenSearchService",
-            new=opensearch_client,
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=mocked_aws_provider,
         ), mock.patch(
             "prowler.providers.aws.services.opensearch.opensearch_service_domains_node_to_node_encryption_enabled.opensearch_service_domains_node_to_node_encryption_enabled.opensearch_client",
-            new=opensearch_client,
+            new=OpenSearchService(mocked_aws_provider),
         ):
             from prowler.providers.aws.services.opensearch.opensearch_service_domains_node_to_node_encryption_enabled.opensearch_service_domains_node_to_node_encryption_enabled import (
                 opensearch_service_domains_node_to_node_encryption_enabled,
@@ -93,8 +102,12 @@ class Test_opensearch_service_domains_node_to_node_encryption_enabled:
             result = check.execute()
             assert len(result) == 1
             assert result[0].status == "PASS"
-            assert search(
-                "has node-to-node encryption enabled", result[0].status_extended
+            assert (
+                result[0].status_extended
+                == "Opensearch domain test-domain-encryption has node-to-node encryption enabled."
+            )
+            assert result[0].resource_id == domain["DomainStatus"]["DomainName"]
+            assert (
+                result[0].resource_arn
+                == f"arn:aws:es:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:domain/{domain['DomainStatus']['DomainName']}"
             )
-            assert result[0].resource_id == domain_name
-            assert result[0].resource_arn == domain_arn

tests/providers/aws/services/opensearch/opensearch_service_domains_not_publicly_accessible/opensearch_service_domains_not_publicly_accessible_test.py

@@ -306,7 +306,7 @@ class Test_opensearch_service_domains_not_publicly_accessible:
 
     def test_domain_inside_vpc(self):
         opensearch_client = mock.MagicMock
-        opensearch_client.opensearch_domains = []
+        opensearch_client.opensearch_domains = {}
 
         aws_provider = set_mocked_aws_provider([AWS_REGION_US_WEST_2])
 
@@ -324,13 +324,12 @@ class Test_opensearch_service_domains_not_publicly_accessible:
                 opensearch_service_domains_not_publicly_accessible,
             )
 
-            opensearch_client.opensearch_domains.append(
-                OpenSearchDomain(
-                    name=domain_name,
-                    region=AWS_REGION_US_WEST_2,
-                    arn=f"arn:aws:es:{AWS_REGION_US_WEST_2}:{AWS_ACCOUNT_NUMBER}:domain/{domain_name}",
-                    vpc_id="vpc-123456",
-                )
+            domain_arn = f"arn:aws:es:{AWS_REGION_US_WEST_2}:{AWS_ACCOUNT_NUMBER}:domain/{domain_name}"
+            opensearch_client.opensearch_domains[domain_arn] = OpenSearchDomain(
+                name=domain_name,
+                region=AWS_REGION_US_WEST_2,
+                arn=domain_arn,
+                vpc_id="vpc-123456",
             )
 
             check = opensearch_service_domains_not_publicly_accessible()

tests/providers/aws/services/opensearch/opensearch_service_domains_updated_to_the_latest_service_software_version/opensearch_service_domains_updated_to_the_latest_service_software_version_test.py

@@ -1,26 +1,76 @@
-from re import search
 from unittest import mock
 
-from prowler.providers.aws.services.opensearch.opensearch_service import (
-    OpenSearchDomain,
-)
-from tests.providers.aws.utils import AWS_ACCOUNT_NUMBER, AWS_REGION_EU_WEST_1
+import botocore
+from boto3 import client
+from moto import mock_aws
 
-domain_name = "test-domain"
-domain_arn = f"arn:aws:es:us-west-2:{AWS_ACCOUNT_NUMBER}:domain/{domain_name}"
+from tests.providers.aws.utils import (
+    AWS_ACCOUNT_NUMBER,
+    AWS_REGION_US_EAST_1,
+    set_mocked_aws_provider,
+)
+
+make_api_call = botocore.client.BaseClient._make_api_call
+
+
+def mock_make_api_call(self, operation_name, kwarg):
+    if operation_name == "ListDomainNames":
+        return {
+            "DomainNames": [
+                {
+                    "DomainName": "test-domain-updates",
+                },
+            ]
+        }
+    if operation_name == "DescribeDomain":
+        return {
+            "DomainStatus": {
+                "DomainName": "test-domain-updates",
+                "EngineVersion": "OpenSearch2.0",
+                "ServiceSoftwareOptions": {
+                    "UpdateAvailable": True,
+                },
+                "ARN": f"arn:aws:es:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:domain/test-domain-updates",
+                "ClusterConfig": {
+                    "InstanceCount": 1,
+                },
+                "AdvancedSecurityOptions": {
+                    "InternalUserDatabaseEnabled": False,
+                },
+                "CognitoOptions": {
+                    "Enabled": False,
+                },
+                "EncryptionAtRestOptions": {
+                    "Enabled": False,
+                },
+                "NodeToNodeEncryptionOptions": {
+                    "Enabled": False,
+                },
+                "DomainEndpointOptions": {
+                    "EnforceHTTPS": False,
+                },
+            }
+        }
+    return make_api_call(self, operation_name, kwarg)
 
 
 class Test_opensearch_service_domains_updated_to_the_latest_service_software_version:
+    @mock_aws
     def test_no_domains(self):
-        opensearch_client = mock.MagicMock
-        opensearch_client.opensearch_domains = []
+        client("opensearch", region_name=AWS_REGION_US_EAST_1)
+
+        from prowler.providers.aws.services.opensearch.opensearch_service import (
+            OpenSearchService,
+        )
+
+        mocked_aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
 
         with mock.patch(
-            "prowler.providers.aws.services.opensearch.opensearch_service.OpenSearchService",
-            new=opensearch_client,
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=mocked_aws_provider,
         ), mock.patch(
             "prowler.providers.aws.services.opensearch.opensearch_service_domains_updated_to_the_latest_service_software_version.opensearch_service_domains_updated_to_the_latest_service_software_version.opensearch_client",
-            new=opensearch_client,
+            new=OpenSearchService(mocked_aws_provider),
         ):
             from prowler.providers.aws.services.opensearch.opensearch_service_domains_updated_to_the_latest_service_software_version.opensearch_service_domains_updated_to_the_latest_service_software_version import (
                 opensearch_service_domains_updated_to_the_latest_service_software_version,
@@ -32,61 +82,22 @@ class Test_opensearch_service_domains_updated_to_the_latest_service_software_ver
             result = check.execute()
             assert len(result) == 0
 
-    def test_internal_update_available(self):
-        opensearch_client = mock.MagicMock
-        opensearch_client.opensearch_domains = []
-        opensearch_client.opensearch_domains.append(
-            OpenSearchDomain(
-                name=domain_name,
-                region=AWS_REGION_EU_WEST_1,
-                arn=domain_arn,
-                update_available=False,
-            )
+    @mock_aws
+    @mock.patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call)
+    def test_updates_available(self):
+        client("opensearch", region_name=AWS_REGION_US_EAST_1)
+        from prowler.providers.aws.services.opensearch.opensearch_service import (
+            OpenSearchService,
         )
-        opensearch_client.opensearch_domains[0].logging = []
+
+        mocked_aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
 
         with mock.patch(
-            "prowler.providers.aws.services.opensearch.opensearch_service.OpenSearchService",
-            new=opensearch_client,
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=mocked_aws_provider,
         ), mock.patch(
             "prowler.providers.aws.services.opensearch.opensearch_service_domains_updated_to_the_latest_service_software_version.opensearch_service_domains_updated_to_the_latest_service_software_version.opensearch_client",
-            new=opensearch_client,
-        ):
-            from prowler.providers.aws.services.opensearch.opensearch_service_domains_updated_to_the_latest_service_software_version.opensearch_service_domains_updated_to_the_latest_service_software_version import (
-                opensearch_service_domains_updated_to_the_latest_service_software_version,
-            )
-
-            check = (
-                opensearch_service_domains_updated_to_the_latest_service_software_version()
-            )
-            result = check.execute()
-            assert len(result) == 1
-            assert result[0].status == "PASS"
-            assert search(
-                "does not have internal updates available", result[0].status_extended
-            )
-            assert result[0].resource_id == domain_name
-            assert result[0].resource_arn == domain_arn
-
-    def test_internal_database_enabled(self):
-        opensearch_client = mock.MagicMock
-        opensearch_client.opensearch_domains = []
-        opensearch_client.opensearch_domains.append(
-            OpenSearchDomain(
-                name=domain_name,
-                region=AWS_REGION_EU_WEST_1,
-                arn=domain_arn,
-                update_available=True,
-            )
-        )
-        opensearch_client.opensearch_domains[0].logging = []
-
-        with mock.patch(
-            "prowler.providers.aws.services.opensearch.opensearch_service.OpenSearchService",
-            new=opensearch_client,
-        ), mock.patch(
-            "prowler.providers.aws.services.opensearch.opensearch_service_domains_updated_to_the_latest_service_software_version.opensearch_service_domains_updated_to_the_latest_service_software_version.opensearch_client",
-            new=opensearch_client,
+            new=OpenSearchService(mocked_aws_provider),
         ):
             from prowler.providers.aws.services.opensearch.opensearch_service_domains_updated_to_the_latest_service_software_version.opensearch_service_domains_updated_to_the_latest_service_software_version import (
                 opensearch_service_domains_updated_to_the_latest_service_software_version,
@@ -98,6 +109,52 @@ class Test_opensearch_service_domains_updated_to_the_latest_service_software_ver
             result = check.execute()
             assert len(result) == 1
             assert result[0].status == "FAIL"
-            assert search("has internal updates available", result[0].status_extended)
-            assert result[0].resource_id == domain_name
-            assert result[0].resource_arn == domain_arn
+            assert (
+                result[0].status_extended
+                == "Opensearch domain test-domain-updates with version OpenSearch2.0 has internal updates available."
+            )
+            assert result[0].resource_id == "test-domain-updates"
+            assert (
+                result[0].resource_arn
+                == f"arn:aws:es:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:domain/test-domain-updates"
+            )
+
+    @mock_aws
+    def test_no_updates_availables(self):
+        opensearch_client = client("opensearch", region_name=AWS_REGION_US_EAST_1)
+        domain = opensearch_client.create_domain(
+            DomainName="test-domain-no-updates",
+            SoftwareUpdateOptions={"AutoSoftwareUpdateEnabled": True},
+        )
+        from prowler.providers.aws.services.opensearch.opensearch_service import (
+            OpenSearchService,
+        )
+
+        mocked_aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=mocked_aws_provider,
+        ), mock.patch(
+            "prowler.providers.aws.services.opensearch.opensearch_service_domains_updated_to_the_latest_service_software_version.opensearch_service_domains_updated_to_the_latest_service_software_version.opensearch_client",
+            new=OpenSearchService(mocked_aws_provider),
+        ):
+            from prowler.providers.aws.services.opensearch.opensearch_service_domains_updated_to_the_latest_service_software_version.opensearch_service_domains_updated_to_the_latest_service_software_version import (
+                opensearch_service_domains_updated_to_the_latest_service_software_version,
+            )
+
+            check = (
+                opensearch_service_domains_updated_to_the_latest_service_software_version()
+            )
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "PASS"
+            assert (
+                result[0].status_extended
+                == f"Opensearch domain test-domain-no-updates with version {domain['DomainStatus']['EngineVersion']} does not have internal updates available."
+            )
+            assert result[0].resource_id == domain["DomainStatus"]["DomainName"]
+            assert (
+                result[0].resource_arn
+                == f"arn:aws:es:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:domain/{domain['DomainStatus']['DomainName']}"
+            )

tests/providers/aws/services/opensearch/opensearch_service_domains_use_cognito_authentication_for_kibana/opensearch_service_domains_use_cognito_authentication_for_kibana_test.py

@@ -1,25 +1,32 @@
 from unittest import mock
 
-from prowler.providers.aws.services.opensearch.opensearch_service import (
-    OpenSearchDomain,
-)
-from tests.providers.aws.utils import AWS_ACCOUNT_NUMBER, AWS_REGION_EU_WEST_1
+from boto3 import client
+from moto import mock_aws
 
-domain_name = "test-domain"
-domain_arn = f"arn:aws:es:us-west-2:{AWS_ACCOUNT_NUMBER}:domain/{domain_name}"
+from tests.providers.aws.utils import (
+    AWS_ACCOUNT_NUMBER,
+    AWS_REGION_US_EAST_1,
+    set_mocked_aws_provider,
+)
 
 
 class Test_opensearch_service_domains_use_cognito_authentication_for_kibana:
+    @mock_aws
     def test_no_domains(self):
-        opensearch_client = mock.MagicMock
-        opensearch_client.opensearch_domains = []
+        client("opensearch", region_name=AWS_REGION_US_EAST_1)
+
+        from prowler.providers.aws.services.opensearch.opensearch_service import (
+            OpenSearchService,
+        )
+
+        mocked_aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
 
         with mock.patch(
-            "prowler.providers.aws.services.opensearch.opensearch_service.OpenSearchService",
-            opensearch_client,
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=mocked_aws_provider,
         ), mock.patch(
             "prowler.providers.aws.services.opensearch.opensearch_service_domains_use_cognito_authentication_for_kibana.opensearch_service_domains_use_cognito_authentication_for_kibana.opensearch_client",
-            new=opensearch_client,
+            new=OpenSearchService(mocked_aws_provider),
         ):
             from prowler.providers.aws.services.opensearch.opensearch_service_domains_use_cognito_authentication_for_kibana.opensearch_service_domains_use_cognito_authentication_for_kibana import (
                 opensearch_service_domains_use_cognito_authentication_for_kibana,
@@ -29,25 +36,24 @@ class Test_opensearch_service_domains_use_cognito_authentication_for_kibana:
             result = check.execute()
             assert len(result) == 0
 
+    @mock_aws
     def test_neither_cognito_nor_saml_enabled(self):
-        opensearch_client = mock.MagicMock
-        opensearch_client.opensearch_domains = []
-        opensearch_client.opensearch_domains.append(
-            OpenSearchDomain(
-                name=domain_name,
-                region=AWS_REGION_EU_WEST_1,
-                arn=domain_arn,
-                cognito_options=False,
-                saml_enabled=False,
-            )
+        opensearch_client = client("opensearch", region_name=AWS_REGION_US_EAST_1)
+        domain = opensearch_client.create_domain(
+            DomainName="test-domain-no-cognito-no-saml",
+        )
+        from prowler.providers.aws.services.opensearch.opensearch_service import (
+            OpenSearchService,
         )
 
+        mocked_aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
+
         with mock.patch(
-            "prowler.providers.aws.services.opensearch.opensearch_service.OpenSearchService",
-            opensearch_client,
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=mocked_aws_provider,
         ), mock.patch(
             "prowler.providers.aws.services.opensearch.opensearch_service_domains_use_cognito_authentication_for_kibana.opensearch_service_domains_use_cognito_authentication_for_kibana.opensearch_client",
-            new=opensearch_client,
+            new=OpenSearchService(mocked_aws_provider),
         ):
             from prowler.providers.aws.services.opensearch.opensearch_service_domains_use_cognito_authentication_for_kibana.opensearch_service_domains_use_cognito_authentication_for_kibana import (
                 opensearch_service_domains_use_cognito_authentication_for_kibana,
@@ -59,31 +65,33 @@ class Test_opensearch_service_domains_use_cognito_authentication_for_kibana:
             assert result[0].status == "FAIL"
             assert (
                 result[0].status_extended
-                == f"Opensearch domain {domain_name} has neither Amazon Cognito nor SAML authentication for Kibana enabled."
+                == "Opensearch domain test-domain-no-cognito-no-saml has neither Amazon Cognito nor SAML authentication for Kibana enabled."
+            )
+            assert result[0].resource_id == domain["DomainStatus"]["DomainName"]
+            assert (
+                result[0].resource_arn
+                == f"arn:aws:es:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:domain/{domain['DomainStatus']['DomainName']}"
             )
-            assert result[0].region == AWS_REGION_EU_WEST_1
-            assert result[0].resource_id == domain_name
-            assert result[0].resource_arn == domain_arn
-            assert result[0].resource_tags == []
 
+    @mock_aws
     def test_cognito_enabled(self):
-        opensearch_client = mock.MagicMock
-        opensearch_client.opensearch_domains = []
-        opensearch_client.opensearch_domains.append(
-            OpenSearchDomain(
-                name=domain_name,
-                region=AWS_REGION_EU_WEST_1,
-                arn=domain_arn,
-                cognito_options=True,
-            )
+        opensearch_client = client("opensearch", region_name=AWS_REGION_US_EAST_1)
+        domain = opensearch_client.create_domain(
+            DomainName="test-domain-cognito",
+            CognitoOptions={"Enabled": True},
+        )
+        from prowler.providers.aws.services.opensearch.opensearch_service import (
+            OpenSearchService,
         )
 
+        mocked_aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
+
         with mock.patch(
-            "prowler.providers.aws.services.opensearch.opensearch_service.OpenSearchService",
-            opensearch_client,
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=mocked_aws_provider,
         ), mock.patch(
             "prowler.providers.aws.services.opensearch.opensearch_service_domains_use_cognito_authentication_for_kibana.opensearch_service_domains_use_cognito_authentication_for_kibana.opensearch_client",
-            new=opensearch_client,
+            new=OpenSearchService(mocked_aws_provider),
         ):
             from prowler.providers.aws.services.opensearch.opensearch_service_domains_use_cognito_authentication_for_kibana.opensearch_service_domains_use_cognito_authentication_for_kibana import (
                 opensearch_service_domains_use_cognito_authentication_for_kibana,
@@ -95,32 +103,33 @@ class Test_opensearch_service_domains_use_cognito_authentication_for_kibana:
             assert result[0].status == "PASS"
             assert (
                 result[0].status_extended
-                == f"Opensearch domain {domain_name} has either Amazon Cognito or SAML authentication for Kibana enabled."
+                == "Opensearch domain test-domain-cognito has either Amazon Cognito or SAML authentication for Kibana enabled."
+            )
+            assert result[0].resource_id == domain["DomainStatus"]["DomainName"]
+            assert (
+                result[0].resource_arn
+                == f"arn:aws:es:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:domain/{domain['DomainStatus']['DomainName']}"
             )
-            assert result[0].region == AWS_REGION_EU_WEST_1
-            assert result[0].resource_id == domain_name
-            assert result[0].resource_arn == domain_arn
-            assert result[0].resource_tags == []
-            assert result[0].resource_arn == domain_arn
 
+    @mock_aws
     def test_saml_enabled(self):
-        opensearch_client = mock.MagicMock
-        opensearch_client.opensearch_domains = []
-        opensearch_client.opensearch_domains.append(
-            OpenSearchDomain(
-                name=domain_name,
-                region=AWS_REGION_EU_WEST_1,
-                arn=domain_arn,
-                saml_enabled=True,
-            )
+        opensearch_client = client("opensearch", region_name=AWS_REGION_US_EAST_1)
+        domain = opensearch_client.create_domain(
+            DomainName="test-domain-saml",
+            AdvancedSecurityOptions={"SAMLOptions": {"Enabled": True}},
+        )
+        from prowler.providers.aws.services.opensearch.opensearch_service import (
+            OpenSearchService,
         )
 
+        mocked_aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
+
         with mock.patch(
-            "prowler.providers.aws.services.opensearch.opensearch_service.OpenSearchService",
-            opensearch_client,
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=mocked_aws_provider,
         ), mock.patch(
             "prowler.providers.aws.services.opensearch.opensearch_service_domains_use_cognito_authentication_for_kibana.opensearch_service_domains_use_cognito_authentication_for_kibana.opensearch_client",
-            new=opensearch_client,
+            new=OpenSearchService(mocked_aws_provider),
         ):
             from prowler.providers.aws.services.opensearch.opensearch_service_domains_use_cognito_authentication_for_kibana.opensearch_service_domains_use_cognito_authentication_for_kibana import (
                 opensearch_service_domains_use_cognito_authentication_for_kibana,
@@ -132,10 +141,10 @@ class Test_opensearch_service_domains_use_cognito_authentication_for_kibana:
             assert result[0].status == "PASS"
             assert (
                 result[0].status_extended
-                == f"Opensearch domain {domain_name} has either Amazon Cognito or SAML authentication for Kibana enabled."
+                == "Opensearch domain test-domain-saml has either Amazon Cognito or SAML authentication for Kibana enabled."
+            )
+            assert result[0].resource_id == domain["DomainStatus"]["DomainName"]
+            assert (
+                result[0].resource_arn
+                == f"arn:aws:es:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:domain/{domain['DomainStatus']['DomainName']}"
             )
-            assert result[0].region == AWS_REGION_EU_WEST_1
-            assert result[0].resource_id == domain_name
-            assert result[0].resource_arn == domain_arn
-            assert result[0].resource_tags == []
-            assert result[0].resource_arn == domain_arn

tests/providers/aws/services/opensearch/opensearch_service_test.py

@@ -2,6 +2,8 @@ from json import dumps
 from unittest.mock import patch
 
 import botocore
+from boto3 import client
+from moto import mock_aws
 
 from prowler.providers.aws.services.opensearch.opensearch_service import (
     OpenSearchService,
@@ -13,7 +15,7 @@ from tests.providers.aws.utils import (
 )
 
 test_domain_name = "test"
-domain_arn = f"arn:aws:es:us-west-2:{AWS_ACCOUNT_NUMBER}:domain/{test_domain_name}"
+domain_arn = f"arn:aws:es:eu-west-1:{AWS_ACCOUNT_NUMBER}:domain/{test_domain_name}"
 
 policy_data = {
     "Version": "2012-10-17",
@@ -56,35 +58,6 @@ def mock_make_api_call(self, operation_name, kwarg):
                 },
             }
         }
-    if operation_name == "DescribeDomain":
-        return {
-            "DomainStatus": {
-                "ARN": domain_arn,
-                "Endpoints": {
-                    "vpc": "vpc-endpoint-h2dsd34efgyghrtguk5gt6j2foh4.us-east-1.es.amazonaws.com"
-                },
-                "EngineVersion": "opensearch-version1",
-                "VPCOptions": {
-                    "VPCId": "test-vpc-id",
-                },
-                "CognitoOptions": {"Enabled": True},
-                "EncryptionAtRestOptions": {"Enabled": True},
-                "NodeToNodeEncryptionOptions": {"Enabled": True},
-                "AdvancedOptions": {"string": "string"},
-                "LogPublishingOptions": {
-                    "string": {
-                        "CloudWatchLogsLogGroupArn": "string",
-                        "Enabled": True | False,
-                    }
-                },
-                "ServiceSoftwareOptions": {"UpdateAvailable": True},
-                "DomainEndpointOptions": {"EnforceHTTPS": True},
-                "AdvancedSecurityOptions": {
-                    "InternalUserDatabaseEnabled": True,
-                    "SAMLOptions": {"Enabled": True},
-                },
-            }
-        }
     if operation_name == "ListTags":
         return {
             "TagList": [
@@ -132,45 +105,80 @@ class Test_OpenSearchService_Service:
         aws_provider = set_mocked_aws_provider([])
         opensearch = OpenSearchService(aws_provider)
         assert len(opensearch.opensearch_domains) == 1
-        assert opensearch.opensearch_domains[0].name == test_domain_name
-        assert opensearch.opensearch_domains[0].region == AWS_REGION_EU_WEST_1
+        assert opensearch.opensearch_domains[domain_arn].name == test_domain_name
+        assert opensearch.opensearch_domains[domain_arn].region == AWS_REGION_EU_WEST_1
 
     # Test OpenSearchService describ domain config
     def test_describe_domain_config(self):
         aws_provider = set_mocked_aws_provider([])
         opensearch = OpenSearchService(aws_provider)
         assert len(opensearch.opensearch_domains) == 1
-        assert opensearch.opensearch_domains[0].name == test_domain_name
-        assert opensearch.opensearch_domains[0].region == AWS_REGION_EU_WEST_1
-        assert opensearch.opensearch_domains[0].access_policy
-        assert opensearch.opensearch_domains[0].logging[0].name == "SEARCH_SLOW_LOGS"
-        assert opensearch.opensearch_domains[0].logging[0].enabled
-        assert opensearch.opensearch_domains[0].logging[1].name == "INDEX_SLOW_LOGS"
-        assert opensearch.opensearch_domains[0].logging[1].enabled
-        assert opensearch.opensearch_domains[0].logging[2].name == "AUDIT_LOGS"
-        assert opensearch.opensearch_domains[0].logging[2].enabled
+        assert opensearch.opensearch_domains[domain_arn].name == test_domain_name
+        assert opensearch.opensearch_domains[domain_arn].region == AWS_REGION_EU_WEST_1
+        assert opensearch.opensearch_domains[domain_arn].access_policy
+        assert (
+            opensearch.opensearch_domains[domain_arn].logging[0].name
+            == "SEARCH_SLOW_LOGS"
+        )
+        assert opensearch.opensearch_domains[domain_arn].logging[0].enabled
+        assert (
+            opensearch.opensearch_domains[domain_arn].logging[1].name
+            == "INDEX_SLOW_LOGS"
+        )
+        assert opensearch.opensearch_domains[domain_arn].logging[1].enabled
+        assert opensearch.opensearch_domains[domain_arn].logging[2].name == "AUDIT_LOGS"
+        assert opensearch.opensearch_domains[domain_arn].logging[2].enabled
 
     # Test OpenSearchService describ domain
+    @mock_aws
     def test_describe_domain(self):
+        opensearch_client = client("opensearch", region_name=AWS_REGION_EU_WEST_1)
+        domain = opensearch_client.create_domain(
+            DomainName=test_domain_name,
+            EncryptionAtRestOptions={"Enabled": True},
+            ClusterConfig={"InstanceCount": 1, "ZoneAwarenessEnabled": True},
+            NodeToNodeEncryptionOptions={"Enabled": True},
+            AdvancedSecurityOptions={
+                "Enabled": True,
+                "InternalUserDatabaseEnabled": True,
+                "SAMLOptions": {"Enabled": True},
+            },
+            CognitoOptions={"Enabled": True},
+            VPCOptions={
+                "SubnetIds": ["test-subnet-id"],
+                "SecurityGroupIds": ["test-security-group-id"],
+            },
+            DomainEndpointOptions={"EnforceHTTPS": True},
+            AccessPolicies=policy_json,
+            EngineVersion="opensearch-version1",
+            TagList=[
+                {"Key": "test", "Value": "test"},
+            ],
+        )
         aws_provider = set_mocked_aws_provider([])
         opensearch = OpenSearchService(aws_provider)
+        domain_arn = domain["DomainStatus"]["ARN"]
         assert len(opensearch.opensearch_domains) == 1
-        assert opensearch.opensearch_domains[0].name == test_domain_name
-        assert opensearch.opensearch_domains[0].region == AWS_REGION_EU_WEST_1
-        assert opensearch.opensearch_domains[0].arn == domain_arn
-        assert opensearch.opensearch_domains[0].access_policy
-        assert opensearch.opensearch_domains[0].vpc_endpoints == [
-            "vpc-endpoint-h2dsd34efgyghrtguk5gt6j2foh4.us-east-1.es.amazonaws.com"
+        assert opensearch.opensearch_domains[domain_arn].name == test_domain_name
+        assert opensearch.opensearch_domains[domain_arn].region == AWS_REGION_EU_WEST_1
+        assert opensearch.opensearch_domains[domain_arn].arn == domain_arn
+        assert opensearch.opensearch_domains[domain_arn].access_policy
+        assert opensearch.opensearch_domains[domain_arn].vpc_endpoints == [
+            "test.eu-west-1.es.amazonaws.com"
         ]
-        assert opensearch.opensearch_domains[0].vpc_id == "test-vpc-id"
-        assert opensearch.opensearch_domains[0].cognito_options
-        assert opensearch.opensearch_domains[0].encryption_at_rest
-        assert opensearch.opensearch_domains[0].node_to_node_encryption
-        assert opensearch.opensearch_domains[0].enforce_https
-        assert opensearch.opensearch_domains[0].internal_user_database
-        assert opensearch.opensearch_domains[0].saml_enabled
-        assert opensearch.opensearch_domains[0].update_available
-        assert opensearch.opensearch_domains[0].version == "opensearch-version1"
-        assert opensearch.opensearch_domains[0].tags == [
+        assert not opensearch.opensearch_domains[domain_arn].vpc_id
+        assert opensearch.opensearch_domains[domain_arn].cognito_options
+        assert opensearch.opensearch_domains[domain_arn].encryption_at_rest
+        assert opensearch.opensearch_domains[domain_arn].node_to_node_encryption
+        assert opensearch.opensearch_domains[domain_arn].enforce_https
+        assert opensearch.opensearch_domains[domain_arn].internal_user_database
+        assert opensearch.opensearch_domains[domain_arn].saml_enabled
+        assert not opensearch.opensearch_domains[domain_arn].update_available
+        assert (
+            opensearch.opensearch_domains[domain_arn].version == "opensearch-version1"
+        )
+        assert opensearch.opensearch_domains[domain_arn].instance_count == 1
+        assert opensearch.opensearch_domains[domain_arn].zone_awareness_enabled
+        assert opensearch.opensearch_domains[domain_arn].tags == [
             {"Key": "test", "Value": "test"},
         ]