diff --git a/prowler/providers/aws/services/opensearch/opensearch_service.py b/prowler/providers/aws/services/opensearch/opensearch_service.py index 5b4c3fb162..b5e284bee6 100644 --- a/prowler/providers/aws/services/opensearch/opensearch_service.py +++ b/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 diff --git a/tests/providers/aws/services/opensearch/opensearch_service_test.py b/tests/providers/aws/services/opensearch/opensearch_service_test.py index 1df689eb18..96d9319c95 100644 --- a/tests/providers/aws/services/opensearch/opensearch_service_test.py +++ b/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"}, ]