From 2bd9f73c1b3a1b285bbd34f5c8018372ca4cda7b Mon Sep 17 00:00:00 2001 From: MarioRgzLpz Date: Thu, 10 Oct 2024 19:12:14 +0200 Subject: [PATCH] refactor(opensearch): Adjust all checks to change from list to dict and change tests to use moto instead of magicMock --- ..._service_domains_access_control_enabled.py | 2 +- ...h_service_domains_audit_logging_enabled.py | 2 +- ...vice_domains_cloudwatch_logging_enabled.py | 2 +- ...vice_domains_encryption_at_rest_enabled.py | 2 +- ...e_domains_https_communications_enforced.py | 2 +- ..._domains_internal_user_database_enabled.py | 2 +- ...domains_node_to_node_encryption_enabled.py | 2 +- ...service_domains_not_publicly_accessible.py | 2 +- ..._to_the_latest_service_software_version.py | 4 +- ...s_use_cognito_authentication_for_kibana.py | 2 +- ...ice_domains_access_control_enabled_test.py | 105 +++++---- ...vice_domains_audit_logging_enabled_test.py | 113 ++++++---- ...domains_cloudwatch_logging_enabled_test.py | 206 +++++++++++------- ...domains_encryption_at_rest_enabled_test.py | 106 +++++---- ...ains_https_communications_enforced_test.py | 110 ++++++---- ...ins_internal_user_database_enabled_test.py | 107 +++++---- ...ns_node_to_node_encryption_enabled_test.py | 107 +++++---- ...ce_domains_not_publicly_accessible_test.py | 15 +- ...he_latest_service_software_version_test.py | 187 ++++++++++------ ..._cognito_authentication_for_kibana_test.py | 139 ++++++------ 20 files changed, 718 insertions(+), 499 deletions(-) diff --git a/prowler/providers/aws/services/opensearch/opensearch_service_domains_access_control_enabled/opensearch_service_domains_access_control_enabled.py b/prowler/providers/aws/services/opensearch/opensearch_service_domains_access_control_enabled/opensearch_service_domains_access_control_enabled.py index 2427be6878..98b9d37e0e 100644 --- a/prowler/providers/aws/services/opensearch/opensearch_service_domains_access_control_enabled/opensearch_service_domains_access_control_enabled.py +++ b/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 diff --git a/prowler/providers/aws/services/opensearch/opensearch_service_domains_audit_logging_enabled/opensearch_service_domains_audit_logging_enabled.py b/prowler/providers/aws/services/opensearch/opensearch_service_domains_audit_logging_enabled/opensearch_service_domains_audit_logging_enabled.py index 3770633d3a..397e110c3a 100644 --- a/prowler/providers/aws/services/opensearch/opensearch_service_domains_audit_logging_enabled/opensearch_service_domains_audit_logging_enabled.py +++ b/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 diff --git a/prowler/providers/aws/services/opensearch/opensearch_service_domains_cloudwatch_logging_enabled/opensearch_service_domains_cloudwatch_logging_enabled.py b/prowler/providers/aws/services/opensearch/opensearch_service_domains_cloudwatch_logging_enabled/opensearch_service_domains_cloudwatch_logging_enabled.py index 1eac7c2cc9..2a6439ab08 100644 --- a/prowler/providers/aws/services/opensearch/opensearch_service_domains_cloudwatch_logging_enabled/opensearch_service_domains_cloudwatch_logging_enabled.py +++ b/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 diff --git a/prowler/providers/aws/services/opensearch/opensearch_service_domains_encryption_at_rest_enabled/opensearch_service_domains_encryption_at_rest_enabled.py b/prowler/providers/aws/services/opensearch/opensearch_service_domains_encryption_at_rest_enabled/opensearch_service_domains_encryption_at_rest_enabled.py index 96c40e5f12..d4e429875e 100644 --- a/prowler/providers/aws/services/opensearch/opensearch_service_domains_encryption_at_rest_enabled/opensearch_service_domains_encryption_at_rest_enabled.py +++ b/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 diff --git a/prowler/providers/aws/services/opensearch/opensearch_service_domains_https_communications_enforced/opensearch_service_domains_https_communications_enforced.py b/prowler/providers/aws/services/opensearch/opensearch_service_domains_https_communications_enforced/opensearch_service_domains_https_communications_enforced.py index 315f27e63e..c352101712 100644 --- a/prowler/providers/aws/services/opensearch/opensearch_service_domains_https_communications_enforced/opensearch_service_domains_https_communications_enforced.py +++ b/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 diff --git a/prowler/providers/aws/services/opensearch/opensearch_service_domains_internal_user_database_enabled/opensearch_service_domains_internal_user_database_enabled.py b/prowler/providers/aws/services/opensearch/opensearch_service_domains_internal_user_database_enabled/opensearch_service_domains_internal_user_database_enabled.py index becb2f3c2e..9d750c2380 100644 --- a/prowler/providers/aws/services/opensearch/opensearch_service_domains_internal_user_database_enabled/opensearch_service_domains_internal_user_database_enabled.py +++ b/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 diff --git a/prowler/providers/aws/services/opensearch/opensearch_service_domains_node_to_node_encryption_enabled/opensearch_service_domains_node_to_node_encryption_enabled.py b/prowler/providers/aws/services/opensearch/opensearch_service_domains_node_to_node_encryption_enabled/opensearch_service_domains_node_to_node_encryption_enabled.py index 4e43f9928e..49efd92ba0 100644 --- a/prowler/providers/aws/services/opensearch/opensearch_service_domains_node_to_node_encryption_enabled/opensearch_service_domains_node_to_node_encryption_enabled.py +++ b/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 diff --git a/prowler/providers/aws/services/opensearch/opensearch_service_domains_not_publicly_accessible/opensearch_service_domains_not_publicly_accessible.py b/prowler/providers/aws/services/opensearch/opensearch_service_domains_not_publicly_accessible/opensearch_service_domains_not_publicly_accessible.py index 8165bf6e4f..3651365cfb 100644 --- a/prowler/providers/aws/services/opensearch/opensearch_service_domains_not_publicly_accessible/opensearch_service_domains_not_publicly_accessible.py +++ b/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 diff --git a/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 b/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 index 6943d29ffd..7c17c9f5b5 100644 --- a/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 +++ b/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." diff --git a/prowler/providers/aws/services/opensearch/opensearch_service_domains_use_cognito_authentication_for_kibana/opensearch_service_domains_use_cognito_authentication_for_kibana.py b/prowler/providers/aws/services/opensearch/opensearch_service_domains_use_cognito_authentication_for_kibana/opensearch_service_domains_use_cognito_authentication_for_kibana.py index cee09fcd41..b3b775685b 100644 --- a/prowler/providers/aws/services/opensearch/opensearch_service_domains_use_cognito_authentication_for_kibana/opensearch_service_domains_use_cognito_authentication_for_kibana.py +++ b/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 diff --git a/tests/providers/aws/services/opensearch/opensearch_service_domains_access_control_enabled/opensearch_service_domains_access_control_enabled_test.py b/tests/providers/aws/services/opensearch/opensearch_service_domains_access_control_enabled/opensearch_service_domains_access_control_enabled_test.py index c679dd1965..a820ffedb3 100644 --- a/tests/providers/aws/services/opensearch/opensearch_service_domains_access_control_enabled/opensearch_service_domains_access_control_enabled_test.py +++ b/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 diff --git a/tests/providers/aws/services/opensearch/opensearch_service_domains_audit_logging_enabled/opensearch_service_domains_audit_logging_enabled_test.py b/tests/providers/aws/services/opensearch/opensearch_service_domains_audit_logging_enabled/opensearch_service_domains_audit_logging_enabled_test.py index 51d4722936..7ef7b22eed 100644 --- a/tests/providers/aws/services/opensearch/opensearch_service_domains_audit_logging_enabled/opensearch_service_domains_audit_logging_enabled_test.py +++ b/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"]}" + ) diff --git a/tests/providers/aws/services/opensearch/opensearch_service_domains_cloudwatch_logging_enabled/opensearch_service_domains_cloudwatch_logging_enabled_test.py b/tests/providers/aws/services/opensearch/opensearch_service_domains_cloudwatch_logging_enabled/opensearch_service_domains_cloudwatch_logging_enabled_test.py index 7454c97916..9cd239fa45 100644 --- a/tests/providers/aws/services/opensearch/opensearch_service_domains_cloudwatch_logging_enabled/opensearch_service_domains_cloudwatch_logging_enabled_test.py +++ b/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 diff --git a/tests/providers/aws/services/opensearch/opensearch_service_domains_encryption_at_rest_enabled/opensearch_service_domains_encryption_at_rest_enabled_test.py b/tests/providers/aws/services/opensearch/opensearch_service_domains_encryption_at_rest_enabled/opensearch_service_domains_encryption_at_rest_enabled_test.py index e50fc0fa5e..431dacc1f4 100644 --- a/tests/providers/aws/services/opensearch/opensearch_service_domains_encryption_at_rest_enabled/opensearch_service_domains_encryption_at_rest_enabled_test.py +++ b/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"]}" + ) diff --git a/tests/providers/aws/services/opensearch/opensearch_service_domains_https_communications_enforced/opensearch_service_domains_https_communications_enforced_test.py b/tests/providers/aws/services/opensearch/opensearch_service_domains_https_communications_enforced/opensearch_service_domains_https_communications_enforced_test.py index 274c52d77a..1a7bb89820 100644 --- a/tests/providers/aws/services/opensearch/opensearch_service_domains_https_communications_enforced/opensearch_service_domains_https_communications_enforced_test.py +++ b/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"]}" + ) diff --git a/tests/providers/aws/services/opensearch/opensearch_service_domains_internal_user_database_enabled/opensearch_service_domains_internal_user_database_enabled_test.py b/tests/providers/aws/services/opensearch/opensearch_service_domains_internal_user_database_enabled/opensearch_service_domains_internal_user_database_enabled_test.py index 10958327b5..fa153d2155 100644 --- a/tests/providers/aws/services/opensearch/opensearch_service_domains_internal_user_database_enabled/opensearch_service_domains_internal_user_database_enabled_test.py +++ b/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 diff --git a/tests/providers/aws/services/opensearch/opensearch_service_domains_node_to_node_encryption_enabled/opensearch_service_domains_node_to_node_encryption_enabled_test.py b/tests/providers/aws/services/opensearch/opensearch_service_domains_node_to_node_encryption_enabled/opensearch_service_domains_node_to_node_encryption_enabled_test.py index fa189dae51..98dd2660fa 100644 --- a/tests/providers/aws/services/opensearch/opensearch_service_domains_node_to_node_encryption_enabled/opensearch_service_domains_node_to_node_encryption_enabled_test.py +++ b/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 diff --git a/tests/providers/aws/services/opensearch/opensearch_service_domains_not_publicly_accessible/opensearch_service_domains_not_publicly_accessible_test.py b/tests/providers/aws/services/opensearch/opensearch_service_domains_not_publicly_accessible/opensearch_service_domains_not_publicly_accessible_test.py index dc6c5514f7..18abd8e30d 100644 --- a/tests/providers/aws/services/opensearch/opensearch_service_domains_not_publicly_accessible/opensearch_service_domains_not_publicly_accessible_test.py +++ b/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() diff --git a/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 b/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 index d4f24699ab..2c664558e7 100644 --- a/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 +++ b/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"]}" + ) diff --git a/tests/providers/aws/services/opensearch/opensearch_service_domains_use_cognito_authentication_for_kibana/opensearch_service_domains_use_cognito_authentication_for_kibana_test.py b/tests/providers/aws/services/opensearch/opensearch_service_domains_use_cognito_authentication_for_kibana/opensearch_service_domains_use_cognito_authentication_for_kibana_test.py index 7e6eddaf55..711b4889a4 100644 --- a/tests/providers/aws/services/opensearch/opensearch_service_domains_use_cognito_authentication_for_kibana/opensearch_service_domains_use_cognito_authentication_for_kibana_test.py +++ b/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