From b6f01b92dd8eeb32e12e9edaa5dd45aa97f99128 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Rub=C3=A9n=20De=20la=20Torre=20Vico?= Date: Wed, 17 Jul 2024 12:03:22 +0200 Subject: [PATCH] test(gcp): Add bigquery and half of cloudsql check tests (#4462) --- .../bigquery_dataset_cmk_encryption_test.py | 105 +++++++++ .../bigquery_table_cmk_encryption_test.py | 104 +++++++++ ...loudsql_instance_automated_backups_test.py | 119 +++++++++++ ...l_instance_mysql_local_infile_flag_test.py | 200 ++++++++++++++++++ ...ance_mysql_skip_show_database_flag_test.py | 200 ++++++++++++++++++ ...tance_postgres_enable_pgaudit_flag_test.py | 200 ++++++++++++++++++ ...ance_postgres_log_connections_flag_test.py | 200 ++++++++++++++++++ ...e_postgres_log_disconnections_flag_test.py | 200 ++++++++++++++++++ ..._postgres_log_error_verbosity_flag_test.py | 200 ++++++++++++++++++ ...es_log_min_duration_statement_flag_test.py | 200 ++++++++++++++++++ ...tgres_log_min_error_statement_flag_test.py | 200 ++++++++++++++++++ ...nce_postgres_log_min_messages_flag_test.py | 200 ++++++++++++++++++ ...stance_postgres_log_statement_flag_test.py | 200 ++++++++++++++++++ 13 files changed, 2328 insertions(+) create mode 100644 tests/providers/gcp/services/bigquery/bigquery_dataset_cmk_encryption/bigquery_dataset_cmk_encryption_test.py create mode 100644 tests/providers/gcp/services/bigquery/bigquery_table_cmk_encryption/bigquery_table_cmk_encryption_test.py create mode 100644 tests/providers/gcp/services/cloudsql/cloudsql_instance_automated_backups/cloudsql_instance_automated_backups_test.py create mode 100644 tests/providers/gcp/services/cloudsql/cloudsql_instance_mysql_local_infile_flag/cloudsql_instance_mysql_local_infile_flag_test.py create mode 100644 tests/providers/gcp/services/cloudsql/cloudsql_instance_mysql_skip_show_database_flag/cloudsql_instance_mysql_skip_show_database_flag_test.py create mode 100644 tests/providers/gcp/services/cloudsql/cloudsql_instance_postgres_enable_pgaudit_flag/cloudsql_instance_postgres_enable_pgaudit_flag_test.py create mode 100644 tests/providers/gcp/services/cloudsql/cloudsql_instance_postgres_log_connections_flag/cloudsql_instance_postgres_log_connections_flag_test.py create mode 100644 tests/providers/gcp/services/cloudsql/cloudsql_instance_postgres_log_disconnections_flag/cloudsql_instance_postgres_log_disconnections_flag_test.py create mode 100644 tests/providers/gcp/services/cloudsql/cloudsql_instance_postgres_log_error_verbosity_flag/cloudsql_instance_postgres_log_error_verbosity_flag_test.py create mode 100644 tests/providers/gcp/services/cloudsql/cloudsql_instance_postgres_log_min_duration_statement_flag/cloudsql_instance_postgres_log_min_duration_statement_flag_test.py create mode 100644 tests/providers/gcp/services/cloudsql/cloudsql_instance_postgres_log_min_error_statement_flag/cloudsql_instance_postgres_log_min_error_statement_flag_test.py create mode 100644 tests/providers/gcp/services/cloudsql/cloudsql_instance_postgres_log_min_messages_flag/cloudsql_instance_postgres_log_min_messages_flag_test.py create mode 100644 tests/providers/gcp/services/cloudsql/cloudsql_instance_postgres_log_statement_flag/cloudsql_instance_postgres_log_statement_flag_test.py diff --git a/tests/providers/gcp/services/bigquery/bigquery_dataset_cmk_encryption/bigquery_dataset_cmk_encryption_test.py b/tests/providers/gcp/services/bigquery/bigquery_dataset_cmk_encryption/bigquery_dataset_cmk_encryption_test.py new file mode 100644 index 0000000000..1eba16d3c1 --- /dev/null +++ b/tests/providers/gcp/services/bigquery/bigquery_dataset_cmk_encryption/bigquery_dataset_cmk_encryption_test.py @@ -0,0 +1,105 @@ +from unittest import mock + +from tests.providers.gcp.gcp_fixtures import GCP_PROJECT_ID, set_mocked_gcp_provider + + +class Test_bigquery_dataset_cmk_encryption: + def test_bigquery_no_datasets(self): + bigquery_client = mock.MagicMock + bigquery_client.datasets = [] + + with mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=set_mocked_gcp_provider(), + ), mock.patch( + "prowler.providers.gcp.services.bigquery.bigquery_dataset_cmk_encryption.bigquery_dataset_cmk_encryption.bigquery_client", + new=bigquery_client, + ): + from prowler.providers.gcp.services.bigquery.bigquery_dataset_cmk_encryption.bigquery_dataset_cmk_encryption import ( + bigquery_dataset_cmk_encryption, + ) + + check = bigquery_dataset_cmk_encryption() + result = check.execute() + assert len(result) == 0 + + def test_one_compliant_dataset(self): + bigquery_client = mock.MagicMock + + with mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=set_mocked_gcp_provider(), + ), mock.patch( + "prowler.providers.gcp.services.bigquery.bigquery_dataset_cmk_encryption.bigquery_dataset_cmk_encryption.bigquery_client", + new=bigquery_client, + ): + from prowler.providers.gcp.services.bigquery.bigquery_dataset_cmk_encryption.bigquery_dataset_cmk_encryption import ( + bigquery_dataset_cmk_encryption, + ) + from prowler.providers.gcp.services.bigquery.bigquery_service import Dataset + + dataset = Dataset( + name="test", + id="1234567890", + region="us-central1", + cmk_encryption=True, + public=False, + project_id=GCP_PROJECT_ID, + ) + bigquery_client.project_ids = [GCP_PROJECT_ID] + bigquery_client.datasets = [dataset] + + check = bigquery_dataset_cmk_encryption() + result = check.execute() + + assert len(result) == 1 + assert result[0].status == "PASS" + assert ( + result[0].status_extended + == f"Dataset {dataset.name} is encrypted with Customer-Managed Keys (CMKs)." + ) + assert result[0].resource_id == dataset.id + assert result[0].resource_name == dataset.name + assert result[0].project_id == dataset.project_id + assert result[0].location == dataset.region + + def test_one_non_compliant_dataset(self): + bigquery_client = mock.MagicMock + + with mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=set_mocked_gcp_provider(), + ), mock.patch( + "prowler.providers.gcp.services.bigquery.bigquery_dataset_cmk_encryption.bigquery_dataset_cmk_encryption.bigquery_client", + new=bigquery_client, + ): + from prowler.providers.gcp.services.bigquery.bigquery_dataset_cmk_encryption.bigquery_dataset_cmk_encryption import ( + bigquery_dataset_cmk_encryption, + ) + from prowler.providers.gcp.services.bigquery.bigquery_service import Dataset + + dataset = Dataset( + name="test", + id="1234567890", + region="us-central1", + cmk_encryption=False, + public=False, + project_id=GCP_PROJECT_ID, + ) + + bigquery_client.project_ids = [GCP_PROJECT_ID] + bigquery_client.datasets = [dataset] + + check = bigquery_dataset_cmk_encryption() + result = check.execute() + + assert len(result) == 1 + assert result[0].status == "FAIL" + assert ( + result[0].status_extended + == f"Dataset {dataset.name} is not encrypted with Customer-Managed Keys (CMKs)." + ) + assert result[0].resource_id == dataset.id + assert result[0].resource_name == dataset.name + assert result[0].project_id == dataset.project_id + assert result[0].location == dataset.region diff --git a/tests/providers/gcp/services/bigquery/bigquery_table_cmk_encryption/bigquery_table_cmk_encryption_test.py b/tests/providers/gcp/services/bigquery/bigquery_table_cmk_encryption/bigquery_table_cmk_encryption_test.py new file mode 100644 index 0000000000..c31de79637 --- /dev/null +++ b/tests/providers/gcp/services/bigquery/bigquery_table_cmk_encryption/bigquery_table_cmk_encryption_test.py @@ -0,0 +1,104 @@ +from unittest import mock + +from tests.providers.gcp.gcp_fixtures import GCP_PROJECT_ID, set_mocked_gcp_provider + + +class Test_bigquery_table_cmk_encryption: + def test_bigquery_no_tables(self): + bigquery_client = mock.MagicMock + bigquery_client.tables = [] + + with mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=set_mocked_gcp_provider(), + ), mock.patch( + "prowler.providers.gcp.services.bigquery.bigquery_table_cmk_encryption.bigquery_table_cmk_encryption.bigquery_client", + new=bigquery_client, + ): + from prowler.providers.gcp.services.bigquery.bigquery_table_cmk_encryption.bigquery_table_cmk_encryption import ( + bigquery_table_cmk_encryption, + ) + + check = bigquery_table_cmk_encryption() + result = check.execute() + assert len(result) == 0 + + def test_one_compliant_table(self): + bigquery_client = mock.MagicMock + + with mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=set_mocked_gcp_provider(), + ), mock.patch( + "prowler.providers.gcp.services.bigquery.bigquery_table_cmk_encryption.bigquery_table_cmk_encryption.bigquery_client", + new=bigquery_client, + ): + from prowler.providers.gcp.services.bigquery.bigquery_service import Table + from prowler.providers.gcp.services.bigquery.bigquery_table_cmk_encryption.bigquery_table_cmk_encryption import ( + bigquery_table_cmk_encryption, + ) + + table = Table( + name="test", + id="1234567890", + region="us-central1", + cmk_encryption=True, + project_id=GCP_PROJECT_ID, + ) + + bigquery_client.project_ids = [GCP_PROJECT_ID] + bigquery_client.tables = [table] + + check = bigquery_table_cmk_encryption() + result = check.execute() + + assert len(result) == 1 + assert result[0].status == "PASS" + assert ( + result[0].status_extended + == f"Table {table.name} is encrypted with Customer-Managed Keys (CMKs)." + ) + assert result[0].resource_id == table.id + assert result[0].resource_name == table.name + assert result[0].project_id == table.project_id + assert result[0].location == table.region + + def test_one_non_compliant_table(self): + bigquery_client = mock.MagicMock + + with mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=set_mocked_gcp_provider(), + ), mock.patch( + "prowler.providers.gcp.services.bigquery.bigquery_table_cmk_encryption.bigquery_table_cmk_encryption.bigquery_client", + new=bigquery_client, + ): + from prowler.providers.gcp.services.bigquery.bigquery_service import Table + from prowler.providers.gcp.services.bigquery.bigquery_table_cmk_encryption.bigquery_table_cmk_encryption import ( + bigquery_table_cmk_encryption, + ) + + table = Table( + name="test", + id="1234567890", + region="us-central1", + cmk_encryption=False, + project_id=GCP_PROJECT_ID, + ) + + bigquery_client.project_ids = [GCP_PROJECT_ID] + bigquery_client.tables = [table] + + check = bigquery_table_cmk_encryption() + result = check.execute() + + assert len(result) == 1 + assert result[0].status == "FAIL" + assert ( + result[0].status_extended + == f"Table {table.name} is not encrypted with Customer-Managed Keys (CMKs)." + ) + assert result[0].resource_id == table.id + assert result[0].resource_name == table.name + assert result[0].project_id == table.project_id + assert result[0].location == table.region diff --git a/tests/providers/gcp/services/cloudsql/cloudsql_instance_automated_backups/cloudsql_instance_automated_backups_test.py b/tests/providers/gcp/services/cloudsql/cloudsql_instance_automated_backups/cloudsql_instance_automated_backups_test.py new file mode 100644 index 0000000000..47fa7e7796 --- /dev/null +++ b/tests/providers/gcp/services/cloudsql/cloudsql_instance_automated_backups/cloudsql_instance_automated_backups_test.py @@ -0,0 +1,119 @@ +from unittest import mock + +from tests.providers.gcp.gcp_fixtures import ( + GCP_EU1_LOCATION, + GCP_PROJECT_ID, + set_mocked_gcp_provider, +) + + +class Test_cloudsql_instance_automated_backups: + def test_no_cloudsql_instances(self): + cloudsql_client = mock.MagicMock + + with mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=set_mocked_gcp_provider(), + ), mock.patch( + "prowler.providers.gcp.services.cloudsql.cloudsql_instance_automated_backups.cloudsql_instance_automated_backups.cloudsql_client", + new=cloudsql_client, + ): + from prowler.providers.gcp.services.cloudsql.cloudsql_instance_automated_backups.cloudsql_instance_automated_backups import ( + cloudsql_instance_automated_backups, + ) + + cloudsql_client.instances = [] + + check = cloudsql_instance_automated_backups() + result = check.execute() + assert len(result) == 0 + + def test_cloudsql_instance_with_automated_backups(self): + cloudsql_client = mock.MagicMock + + with mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=set_mocked_gcp_provider(), + ), mock.patch( + "prowler.providers.gcp.services.cloudsql.cloudsql_instance_automated_backups.cloudsql_instance_automated_backups.cloudsql_client", + new=cloudsql_client, + ): + from prowler.providers.gcp.services.cloudsql.cloudsql_instance_automated_backups.cloudsql_instance_automated_backups import ( + cloudsql_instance_automated_backups, + ) + from prowler.providers.gcp.services.cloudsql.cloudsql_service import ( + Instance, + ) + + cloudsql_client.instances = [ + Instance( + name="instance1", + version="POSTGRES_15", + ip_addresses=[], + region=GCP_EU1_LOCATION, + public_ip=False, + ssl=False, + automated_backups=True, + authorized_networks=[], + flags=[], + project_id=GCP_PROJECT_ID, + ) + ] + + check = cloudsql_instance_automated_backups() + result = check.execute() + assert len(result) == 1 + assert result[0].status == "PASS" + assert ( + result[0].status_extended + == "Database Instance instance1 has automated backups configured." + ) + assert result[0].resource_id == "instance1" + assert result[0].resource_name == "instance1" + assert result[0].location == GCP_EU1_LOCATION + assert result[0].project_id == GCP_PROJECT_ID + + def test_cloudsql_instance_without_automated_backups(self): + cloudsql_client = mock.MagicMock + + with mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=set_mocked_gcp_provider(), + ), mock.patch( + "prowler.providers.gcp.services.cloudsql.cloudsql_instance_automated_backups.cloudsql_instance_automated_backups.cloudsql_client", + new=cloudsql_client, + ): + from prowler.providers.gcp.services.cloudsql.cloudsql_instance_automated_backups.cloudsql_instance_automated_backups import ( + cloudsql_instance_automated_backups, + ) + from prowler.providers.gcp.services.cloudsql.cloudsql_service import ( + Instance, + ) + + cloudsql_client.instances = [ + Instance( + name="instance1", + version="POSTGRES_15", + ip_addresses=[], + region=GCP_EU1_LOCATION, + public_ip=False, + ssl=False, + automated_backups=False, + authorized_networks=[], + flags=[], + project_id=GCP_PROJECT_ID, + ) + ] + + check = cloudsql_instance_automated_backups() + result = check.execute() + assert len(result) == 1 + assert result[0].status == "FAIL" + assert ( + result[0].status_extended + == "Database Instance instance1 does not have automated backups configured." + ) + assert result[0].resource_id == "instance1" + assert result[0].resource_name == "instance1" + assert result[0].location == GCP_EU1_LOCATION + assert result[0].project_id == GCP_PROJECT_ID diff --git a/tests/providers/gcp/services/cloudsql/cloudsql_instance_mysql_local_infile_flag/cloudsql_instance_mysql_local_infile_flag_test.py b/tests/providers/gcp/services/cloudsql/cloudsql_instance_mysql_local_infile_flag/cloudsql_instance_mysql_local_infile_flag_test.py new file mode 100644 index 0000000000..e1b3a156b8 --- /dev/null +++ b/tests/providers/gcp/services/cloudsql/cloudsql_instance_mysql_local_infile_flag/cloudsql_instance_mysql_local_infile_flag_test.py @@ -0,0 +1,200 @@ +from unittest import mock + +from tests.providers.gcp.gcp_fixtures import ( + GCP_EU1_LOCATION, + GCP_PROJECT_ID, + set_mocked_gcp_provider, +) + + +class Test_cloudsql_instance_mysql_local_infile_flag: + def test_no_cloudsql_instances(self): + cloudsql_client = mock.MagicMock + + with mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=set_mocked_gcp_provider(), + ), mock.patch( + "prowler.providers.gcp.services.cloudsql.cloudsql_instance_mysql_local_infile_flag.cloudsql_instance_mysql_local_infile_flag.cloudsql_client", + new=cloudsql_client, + ): + from prowler.providers.gcp.services.cloudsql.cloudsql_instance_mysql_local_infile_flag.cloudsql_instance_mysql_local_infile_flag import ( + cloudsql_instance_mysql_local_infile_flag, + ) + + cloudsql_client.instances = [] + + check = cloudsql_instance_mysql_local_infile_flag() + result = check.execute() + assert len(result) == 0 + + def test_cloudsql_postgres_instance(self): + cloudsql_client = mock.MagicMock + + with mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=set_mocked_gcp_provider(), + ), mock.patch( + "prowler.providers.gcp.services.cloudsql.cloudsql_instance_mysql_local_infile_flag.cloudsql_instance_mysql_local_infile_flag.cloudsql_client", + new=cloudsql_client, + ): + from prowler.providers.gcp.services.cloudsql.cloudsql_instance_mysql_local_infile_flag.cloudsql_instance_mysql_local_infile_flag import ( + cloudsql_instance_mysql_local_infile_flag, + ) + from prowler.providers.gcp.services.cloudsql.cloudsql_service import ( + Instance, + ) + + cloudsql_client.instances = [ + Instance( + name="instance1", + version="POSTGRES_15", + ip_addresses=[], + region=GCP_EU1_LOCATION, + public_ip=False, + ssl=False, + automated_backups=True, + authorized_networks=[], + flags=[], + project_id=GCP_PROJECT_ID, + ) + ] + + check = cloudsql_instance_mysql_local_infile_flag() + result = check.execute() + assert len(result) == 0 + + def test_cloudsql_instance_with_no_flags(self): + cloudsql_client = mock.MagicMock + + with mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=set_mocked_gcp_provider(), + ), mock.patch( + "prowler.providers.gcp.services.cloudsql.cloudsql_instance_mysql_local_infile_flag.cloudsql_instance_mysql_local_infile_flag.cloudsql_client", + new=cloudsql_client, + ): + from prowler.providers.gcp.services.cloudsql.cloudsql_instance_mysql_local_infile_flag.cloudsql_instance_mysql_local_infile_flag import ( + cloudsql_instance_mysql_local_infile_flag, + ) + from prowler.providers.gcp.services.cloudsql.cloudsql_service import ( + Instance, + ) + + cloudsql_client.instances = [ + Instance( + name="instance1", + version="MYSQL_15", + ip_addresses=[], + region=GCP_EU1_LOCATION, + public_ip=False, + ssl=False, + automated_backups=True, + authorized_networks=[], + flags=[], + project_id=GCP_PROJECT_ID, + ) + ] + + check = cloudsql_instance_mysql_local_infile_flag() + result = check.execute() + assert len(result) == 1 + assert result[0].status == "FAIL" + assert ( + result[0].status_extended + == "MySQL Instance instance1 does not have 'local_infile' flag set to 'off'." + ) + assert result[0].resource_id == "instance1" + assert result[0].resource_name == "instance1" + assert result[0].location == GCP_EU1_LOCATION + assert result[0].project_id == GCP_PROJECT_ID + + def test_cloudsql_instance_with_local_infile_flag_off(self): + cloudsql_client = mock.MagicMock + + with mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=set_mocked_gcp_provider(), + ), mock.patch( + "prowler.providers.gcp.services.cloudsql.cloudsql_instance_mysql_local_infile_flag.cloudsql_instance_mysql_local_infile_flag.cloudsql_client", + new=cloudsql_client, + ): + from prowler.providers.gcp.services.cloudsql.cloudsql_instance_mysql_local_infile_flag.cloudsql_instance_mysql_local_infile_flag import ( + cloudsql_instance_mysql_local_infile_flag, + ) + from prowler.providers.gcp.services.cloudsql.cloudsql_service import ( + Instance, + ) + + cloudsql_client.instances = [ + Instance( + name="instance1", + version="MYSQL_15", + ip_addresses=[], + region=GCP_EU1_LOCATION, + public_ip=False, + ssl=False, + automated_backups=True, + authorized_networks=[], + flags=[{"name": "local_infile", "value": "off"}], + project_id=GCP_PROJECT_ID, + ) + ] + + check = cloudsql_instance_mysql_local_infile_flag() + result = check.execute() + assert len(result) == 1 + assert result[0].status == "PASS" + assert ( + result[0].status_extended + == "MySQL Instance instance1 has 'local_infile' flag set to 'off'." + ) + assert result[0].resource_id == "instance1" + assert result[0].resource_name == "instance1" + assert result[0].location == GCP_EU1_LOCATION + assert result[0].project_id == GCP_PROJECT_ID + + def test_cloudsql_instance_with_local_infile_flag_on(self): + cloudsql_client = mock.MagicMock + + with mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=set_mocked_gcp_provider(), + ), mock.patch( + "prowler.providers.gcp.services.cloudsql.cloudsql_instance_mysql_local_infile_flag.cloudsql_instance_mysql_local_infile_flag.cloudsql_client", + new=cloudsql_client, + ): + from prowler.providers.gcp.services.cloudsql.cloudsql_instance_mysql_local_infile_flag.cloudsql_instance_mysql_local_infile_flag import ( + cloudsql_instance_mysql_local_infile_flag, + ) + from prowler.providers.gcp.services.cloudsql.cloudsql_service import ( + Instance, + ) + + cloudsql_client.instances = [ + Instance( + name="instance1", + version="MYSQL_15", + ip_addresses=[], + region=GCP_EU1_LOCATION, + public_ip=False, + ssl=False, + automated_backups=True, + authorized_networks=[], + flags=[{"name": "local_infile", "value": "on"}], + project_id=GCP_PROJECT_ID, + ) + ] + + check = cloudsql_instance_mysql_local_infile_flag() + result = check.execute() + assert len(result) == 1 + assert result[0].status == "FAIL" + assert ( + result[0].status_extended + == "MySQL Instance instance1 does not have 'local_infile' flag set to 'off'." + ) + assert result[0].resource_id == "instance1" + assert result[0].resource_name == "instance1" + assert result[0].location == GCP_EU1_LOCATION + assert result[0].project_id == GCP_PROJECT_ID diff --git a/tests/providers/gcp/services/cloudsql/cloudsql_instance_mysql_skip_show_database_flag/cloudsql_instance_mysql_skip_show_database_flag_test.py b/tests/providers/gcp/services/cloudsql/cloudsql_instance_mysql_skip_show_database_flag/cloudsql_instance_mysql_skip_show_database_flag_test.py new file mode 100644 index 0000000000..2615b7380a --- /dev/null +++ b/tests/providers/gcp/services/cloudsql/cloudsql_instance_mysql_skip_show_database_flag/cloudsql_instance_mysql_skip_show_database_flag_test.py @@ -0,0 +1,200 @@ +from unittest import mock + +from tests.providers.gcp.gcp_fixtures import ( + GCP_EU1_LOCATION, + GCP_PROJECT_ID, + set_mocked_gcp_provider, +) + + +class Test_cloudsql_instance_mysql_skip_show_database_flag: + def test_no_cloudsql_instances(self): + cloudsql_client = mock.MagicMock + + with mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=set_mocked_gcp_provider(), + ), mock.patch( + "prowler.providers.gcp.services.cloudsql.cloudsql_instance_mysql_skip_show_database_flag.cloudsql_instance_mysql_skip_show_database_flag.cloudsql_client", + new=cloudsql_client, + ): + from prowler.providers.gcp.services.cloudsql.cloudsql_instance_mysql_skip_show_database_flag.cloudsql_instance_mysql_skip_show_database_flag import ( + cloudsql_instance_mysql_skip_show_database_flag, + ) + + cloudsql_client.instances = [] + + check = cloudsql_instance_mysql_skip_show_database_flag() + result = check.execute() + assert len(result) == 0 + + def test_cloudsql_postgres_instance(self): + cloudsql_client = mock.MagicMock + + with mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=set_mocked_gcp_provider(), + ), mock.patch( + "prowler.providers.gcp.services.cloudsql.cloudsql_instance_mysql_skip_show_database_flag.cloudsql_instance_mysql_skip_show_database_flag.cloudsql_client", + new=cloudsql_client, + ): + from prowler.providers.gcp.services.cloudsql.cloudsql_instance_mysql_skip_show_database_flag.cloudsql_instance_mysql_skip_show_database_flag import ( + cloudsql_instance_mysql_skip_show_database_flag, + ) + from prowler.providers.gcp.services.cloudsql.cloudsql_service import ( + Instance, + ) + + cloudsql_client.instances = [ + Instance( + name="instance1", + version="POSTGRES_15", + ip_addresses=[], + region=GCP_EU1_LOCATION, + public_ip=False, + ssl=False, + automated_backups=True, + authorized_networks=[], + flags=[], + project_id=GCP_PROJECT_ID, + ) + ] + + check = cloudsql_instance_mysql_skip_show_database_flag() + result = check.execute() + assert len(result) == 0 + + def test_cloudsql_instance_no_flags(self): + cloudsql_client = mock.MagicMock + + with mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=set_mocked_gcp_provider(), + ), mock.patch( + "prowler.providers.gcp.services.cloudsql.cloudsql_instance_mysql_skip_show_database_flag.cloudsql_instance_mysql_skip_show_database_flag.cloudsql_client", + new=cloudsql_client, + ): + from prowler.providers.gcp.services.cloudsql.cloudsql_instance_mysql_skip_show_database_flag.cloudsql_instance_mysql_skip_show_database_flag import ( + cloudsql_instance_mysql_skip_show_database_flag, + ) + from prowler.providers.gcp.services.cloudsql.cloudsql_service import ( + Instance, + ) + + cloudsql_client.instances = [ + Instance( + name="instance1", + version="MYSQL_15", + ip_addresses=[], + region=GCP_EU1_LOCATION, + public_ip=False, + ssl=False, + automated_backups=True, + authorized_networks=[], + flags=[], + project_id=GCP_PROJECT_ID, + ) + ] + + check = cloudsql_instance_mysql_skip_show_database_flag() + result = check.execute() + assert len(result) == 1 + assert result[0].status == "FAIL" + assert ( + result[0].status_extended + == "MySQL Instance instance1 does not have 'skip_show_database' flag set to 'on'." + ) + assert result[0].resource_id == "instance1" + assert result[0].resource_name == "instance1" + assert result[0].location == GCP_EU1_LOCATION + assert result[0].project_id == GCP_PROJECT_ID + + def test_cloudsql_instance_with_skip_show_databases_off(self): + cloudsql_client = mock.MagicMock + + with mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=set_mocked_gcp_provider(), + ), mock.patch( + "prowler.providers.gcp.services.cloudsql.cloudsql_instance_mysql_skip_show_database_flag.cloudsql_instance_mysql_skip_show_database_flag.cloudsql_client", + new=cloudsql_client, + ): + from prowler.providers.gcp.services.cloudsql.cloudsql_instance_mysql_skip_show_database_flag.cloudsql_instance_mysql_skip_show_database_flag import ( + cloudsql_instance_mysql_skip_show_database_flag, + ) + from prowler.providers.gcp.services.cloudsql.cloudsql_service import ( + Instance, + ) + + cloudsql_client.instances = [ + Instance( + name="instance1", + version="MYSQL_15", + ip_addresses=[], + region=GCP_EU1_LOCATION, + public_ip=False, + ssl=False, + automated_backups=True, + authorized_networks=[], + flags=[{"name": "skip_show_database", "value": "off"}], + project_id=GCP_PROJECT_ID, + ) + ] + + check = cloudsql_instance_mysql_skip_show_database_flag() + result = check.execute() + assert len(result) == 1 + assert result[0].status == "FAIL" + assert ( + result[0].status_extended + == "MySQL Instance instance1 does not have 'skip_show_database' flag set to 'on'." + ) + assert result[0].resource_id == "instance1" + assert result[0].resource_name == "instance1" + assert result[0].location == GCP_EU1_LOCATION + assert result[0].project_id == GCP_PROJECT_ID + + def test_cloudsql_instance_with_skip_show_databases_on(self): + cloudsql_client = mock.MagicMock + + with mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=set_mocked_gcp_provider(), + ), mock.patch( + "prowler.providers.gcp.services.cloudsql.cloudsql_instance_mysql_skip_show_database_flag.cloudsql_instance_mysql_skip_show_database_flag.cloudsql_client", + new=cloudsql_client, + ): + from prowler.providers.gcp.services.cloudsql.cloudsql_instance_mysql_skip_show_database_flag.cloudsql_instance_mysql_skip_show_database_flag import ( + cloudsql_instance_mysql_skip_show_database_flag, + ) + from prowler.providers.gcp.services.cloudsql.cloudsql_service import ( + Instance, + ) + + cloudsql_client.instances = [ + Instance( + name="instance1", + version="MYSQL_15", + ip_addresses=[], + region=GCP_EU1_LOCATION, + public_ip=False, + ssl=False, + automated_backups=True, + authorized_networks=[], + flags=[{"name": "skip_show_database", "value": "on"}], + project_id=GCP_PROJECT_ID, + ) + ] + + check = cloudsql_instance_mysql_skip_show_database_flag() + result = check.execute() + assert len(result) == 1 + assert result[0].status == "PASS" + assert ( + result[0].status_extended + == "MySQL Instance instance1 has 'skip_show_database' flag set to 'on'." + ) + assert result[0].resource_id == "instance1" + assert result[0].resource_name == "instance1" + assert result[0].location == GCP_EU1_LOCATION + assert result[0].project_id == GCP_PROJECT_ID diff --git a/tests/providers/gcp/services/cloudsql/cloudsql_instance_postgres_enable_pgaudit_flag/cloudsql_instance_postgres_enable_pgaudit_flag_test.py b/tests/providers/gcp/services/cloudsql/cloudsql_instance_postgres_enable_pgaudit_flag/cloudsql_instance_postgres_enable_pgaudit_flag_test.py new file mode 100644 index 0000000000..97a9f8aa01 --- /dev/null +++ b/tests/providers/gcp/services/cloudsql/cloudsql_instance_postgres_enable_pgaudit_flag/cloudsql_instance_postgres_enable_pgaudit_flag_test.py @@ -0,0 +1,200 @@ +from unittest import mock + +from tests.providers.gcp.gcp_fixtures import ( + GCP_EU1_LOCATION, + GCP_PROJECT_ID, + set_mocked_gcp_provider, +) + + +class Test_cloudsql_instance_postgres_enable_pgaudit_flag: + def test_no_cloudsql_instances(self): + cloudsql_client = mock.MagicMock + + with mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=set_mocked_gcp_provider(), + ), mock.patch( + "prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_enable_pgaudit_flag.cloudsql_instance_postgres_enable_pgaudit_flag.cloudsql_client", + new=cloudsql_client, + ): + from prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_enable_pgaudit_flag.cloudsql_instance_postgres_enable_pgaudit_flag import ( + cloudsql_instance_postgres_enable_pgaudit_flag, + ) + + cloudsql_client.instances = [] + + check = cloudsql_instance_postgres_enable_pgaudit_flag() + result = check.execute() + assert len(result) == 0 + + def test_cloudsql_mysql_instance(self): + cloudsql_client = mock.MagicMock + + with mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=set_mocked_gcp_provider(), + ), mock.patch( + "prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_enable_pgaudit_flag.cloudsql_instance_postgres_enable_pgaudit_flag.cloudsql_client", + new=cloudsql_client, + ): + from prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_enable_pgaudit_flag.cloudsql_instance_postgres_enable_pgaudit_flag import ( + cloudsql_instance_postgres_enable_pgaudit_flag, + ) + from prowler.providers.gcp.services.cloudsql.cloudsql_service import ( + Instance, + ) + + cloudsql_client.instances = [ + Instance( + name="instance1", + version="MYSQL_5_7", + ip_addresses=[], + region=GCP_EU1_LOCATION, + public_ip=False, + ssl=False, + automated_backups=True, + authorized_networks=[], + flags=[], + project_id=GCP_PROJECT_ID, + ) + ] + + check = cloudsql_instance_postgres_enable_pgaudit_flag() + result = check.execute() + assert len(result) == 0 + + def test_cloudsql_instance_no_flags(self): + cloudsql_client = mock.MagicMock + + with mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=set_mocked_gcp_provider(), + ), mock.patch( + "prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_enable_pgaudit_flag.cloudsql_instance_postgres_enable_pgaudit_flag.cloudsql_client", + new=cloudsql_client, + ): + from prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_enable_pgaudit_flag.cloudsql_instance_postgres_enable_pgaudit_flag import ( + cloudsql_instance_postgres_enable_pgaudit_flag, + ) + from prowler.providers.gcp.services.cloudsql.cloudsql_service import ( + Instance, + ) + + cloudsql_client.instances = [ + Instance( + name="instance1", + version="POSTGRES_15", + ip_addresses=[], + region=GCP_EU1_LOCATION, + public_ip=False, + ssl=False, + automated_backups=True, + authorized_networks=[], + flags=[], + project_id=GCP_PROJECT_ID, + ) + ] + + check = cloudsql_instance_postgres_enable_pgaudit_flag() + result = check.execute() + assert len(result) == 1 + assert result[0].status == "FAIL" + assert ( + result[0].status_extended + == "PostgreSQL Instance instance1 does not have 'cloudsql.enable_pgaudit' flag set to 'on'." + ) + assert result[0].resource_id == "instance1" + assert result[0].resource_name == "instance1" + assert result[0].location == GCP_EU1_LOCATION + assert result[0].project_id == GCP_PROJECT_ID + + def test_cloudsql_instance_pgaudit_flag_off(self): + cloudsql_client = mock.MagicMock + + with mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=set_mocked_gcp_provider(), + ), mock.patch( + "prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_enable_pgaudit_flag.cloudsql_instance_postgres_enable_pgaudit_flag.cloudsql_client", + new=cloudsql_client, + ): + from prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_enable_pgaudit_flag.cloudsql_instance_postgres_enable_pgaudit_flag import ( + cloudsql_instance_postgres_enable_pgaudit_flag, + ) + from prowler.providers.gcp.services.cloudsql.cloudsql_service import ( + Instance, + ) + + cloudsql_client.instances = [ + Instance( + name="instance1", + version="POSTGRES_15", + ip_addresses=[], + region=GCP_EU1_LOCATION, + public_ip=False, + ssl=False, + automated_backups=True, + authorized_networks=[], + flags=[{"name": "cloudsql.enable_pgaudit", "value": "off"}], + project_id=GCP_PROJECT_ID, + ) + ] + + check = cloudsql_instance_postgres_enable_pgaudit_flag() + result = check.execute() + assert len(result) == 1 + assert result[0].status == "FAIL" + assert ( + result[0].status_extended + == "PostgreSQL Instance instance1 does not have 'cloudsql.enable_pgaudit' flag set to 'on'." + ) + assert result[0].resource_id == "instance1" + assert result[0].resource_name == "instance1" + assert result[0].location == GCP_EU1_LOCATION + assert result[0].project_id == GCP_PROJECT_ID + + def test_cloudsql_instance_pgaudit_flag_on(self): + cloudsql_client = mock.MagicMock + + with mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=set_mocked_gcp_provider(), + ), mock.patch( + "prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_enable_pgaudit_flag.cloudsql_instance_postgres_enable_pgaudit_flag.cloudsql_client", + new=cloudsql_client, + ): + from prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_enable_pgaudit_flag.cloudsql_instance_postgres_enable_pgaudit_flag import ( + cloudsql_instance_postgres_enable_pgaudit_flag, + ) + from prowler.providers.gcp.services.cloudsql.cloudsql_service import ( + Instance, + ) + + cloudsql_client.instances = [ + Instance( + name="instance1", + version="POSTGRES_15", + ip_addresses=[], + region=GCP_EU1_LOCATION, + public_ip=False, + ssl=False, + automated_backups=True, + authorized_networks=[], + flags=[{"name": "cloudsql.enable_pgaudit", "value": "on"}], + project_id=GCP_PROJECT_ID, + ) + ] + + check = cloudsql_instance_postgres_enable_pgaudit_flag() + result = check.execute() + assert len(result) == 1 + assert result[0].status == "PASS" + assert ( + result[0].status_extended + == "PostgreSQL Instance instance1 has 'cloudsql.enable_pgaudit' flag set to 'on'." + ) + assert result[0].resource_id == "instance1" + assert result[0].resource_name == "instance1" + assert result[0].location == GCP_EU1_LOCATION + assert result[0].project_id == GCP_PROJECT_ID diff --git a/tests/providers/gcp/services/cloudsql/cloudsql_instance_postgres_log_connections_flag/cloudsql_instance_postgres_log_connections_flag_test.py b/tests/providers/gcp/services/cloudsql/cloudsql_instance_postgres_log_connections_flag/cloudsql_instance_postgres_log_connections_flag_test.py new file mode 100644 index 0000000000..e1dfe603de --- /dev/null +++ b/tests/providers/gcp/services/cloudsql/cloudsql_instance_postgres_log_connections_flag/cloudsql_instance_postgres_log_connections_flag_test.py @@ -0,0 +1,200 @@ +from unittest import mock + +from tests.providers.gcp.gcp_fixtures import ( + GCP_EU1_LOCATION, + GCP_PROJECT_ID, + set_mocked_gcp_provider, +) + + +class Test_cloudsql_instance_postgres_log_connections_flag: + def test_no_cloudsql_instances(self): + cloudsql_client = mock.MagicMock + + with mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=set_mocked_gcp_provider(), + ), mock.patch( + "prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_connections_flag.cloudsql_instance_postgres_log_connections_flag.cloudsql_client", + new=cloudsql_client, + ): + from prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_connections_flag.cloudsql_instance_postgres_log_connections_flag import ( + cloudsql_instance_postgres_log_connections_flag, + ) + + cloudsql_client.instances = [] + + check = cloudsql_instance_postgres_log_connections_flag() + result = check.execute() + assert len(result) == 0 + + def test_cloudsql_mysql_instance(self): + cloudsql_client = mock.MagicMock + + with mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=set_mocked_gcp_provider(), + ), mock.patch( + "prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_connections_flag.cloudsql_instance_postgres_log_connections_flag.cloudsql_client", + new=cloudsql_client, + ): + from prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_connections_flag.cloudsql_instance_postgres_log_connections_flag import ( + cloudsql_instance_postgres_log_connections_flag, + ) + from prowler.providers.gcp.services.cloudsql.cloudsql_service import ( + Instance, + ) + + cloudsql_client.instances = [ + Instance( + name="instance1", + version="MYSQL_5_7", + ip_addresses=[], + region=GCP_EU1_LOCATION, + public_ip=False, + ssl=False, + automated_backups=True, + authorized_networks=[], + flags=[], + project_id=GCP_PROJECT_ID, + ) + ] + + check = cloudsql_instance_postgres_log_connections_flag() + result = check.execute() + assert len(result) == 0 + + def test_cloudsql_instance_no_flags(self): + cloudsql_client = mock.MagicMock + + with mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=set_mocked_gcp_provider(), + ), mock.patch( + "prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_connections_flag.cloudsql_instance_postgres_log_connections_flag.cloudsql_client", + new=cloudsql_client, + ): + from prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_connections_flag.cloudsql_instance_postgres_log_connections_flag import ( + cloudsql_instance_postgres_log_connections_flag, + ) + from prowler.providers.gcp.services.cloudsql.cloudsql_service import ( + Instance, + ) + + cloudsql_client.instances = [ + Instance( + name="instance1", + version="POSTGRES_15", + ip_addresses=[], + region=GCP_EU1_LOCATION, + public_ip=False, + ssl=False, + automated_backups=True, + authorized_networks=[], + flags=[], + project_id=GCP_PROJECT_ID, + ) + ] + + check = cloudsql_instance_postgres_log_connections_flag() + result = check.execute() + assert len(result) == 1 + assert result[0].status == "FAIL" + assert ( + result[0].status_extended + == "PostgreSQL Instance instance1 does not have 'log_connections' flag set to 'on'." + ) + assert result[0].resource_id == "instance1" + assert result[0].resource_name == "instance1" + assert result[0].location == GCP_EU1_LOCATION + assert result[0].project_id == GCP_PROJECT_ID + + def test_cloudsql_instance_log_connections_flag_off(self): + cloudsql_client = mock.MagicMock + + with mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=set_mocked_gcp_provider(), + ), mock.patch( + "prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_connections_flag.cloudsql_instance_postgres_log_connections_flag.cloudsql_client", + new=cloudsql_client, + ): + from prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_connections_flag.cloudsql_instance_postgres_log_connections_flag import ( + cloudsql_instance_postgres_log_connections_flag, + ) + from prowler.providers.gcp.services.cloudsql.cloudsql_service import ( + Instance, + ) + + cloudsql_client.instances = [ + Instance( + name="instance1", + version="POSTGRES_15", + ip_addresses=[], + region=GCP_EU1_LOCATION, + public_ip=False, + ssl=False, + automated_backups=True, + authorized_networks=[], + flags=[{"name": "log_connections", "value": "off"}], + project_id=GCP_PROJECT_ID, + ) + ] + + check = cloudsql_instance_postgres_log_connections_flag() + result = check.execute() + assert len(result) == 1 + assert result[0].status == "FAIL" + assert ( + result[0].status_extended + == "PostgreSQL Instance instance1 does not have 'log_connections' flag set to 'on'." + ) + assert result[0].resource_id == "instance1" + assert result[0].resource_name == "instance1" + assert result[0].location == GCP_EU1_LOCATION + assert result[0].project_id == GCP_PROJECT_ID + + def test_cloudsql_instance_log_connections_flag_on(self): + cloudsql_client = mock.MagicMock + + with mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=set_mocked_gcp_provider(), + ), mock.patch( + "prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_connections_flag.cloudsql_instance_postgres_log_connections_flag.cloudsql_client", + new=cloudsql_client, + ): + from prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_connections_flag.cloudsql_instance_postgres_log_connections_flag import ( + cloudsql_instance_postgres_log_connections_flag, + ) + from prowler.providers.gcp.services.cloudsql.cloudsql_service import ( + Instance, + ) + + cloudsql_client.instances = [ + Instance( + name="instance1", + version="POSTGRES_15", + ip_addresses=[], + region=GCP_EU1_LOCATION, + public_ip=False, + ssl=False, + automated_backups=True, + authorized_networks=[], + flags=[{"name": "log_connections", "value": "on"}], + project_id=GCP_PROJECT_ID, + ) + ] + + check = cloudsql_instance_postgres_log_connections_flag() + result = check.execute() + assert len(result) == 1 + assert result[0].status == "PASS" + assert ( + result[0].status_extended + == "PostgreSQL Instance instance1 has 'log_connections' flag set to 'on'." + ) + assert result[0].resource_id == "instance1" + assert result[0].resource_name == "instance1" + assert result[0].location == GCP_EU1_LOCATION + assert result[0].project_id == GCP_PROJECT_ID diff --git a/tests/providers/gcp/services/cloudsql/cloudsql_instance_postgres_log_disconnections_flag/cloudsql_instance_postgres_log_disconnections_flag_test.py b/tests/providers/gcp/services/cloudsql/cloudsql_instance_postgres_log_disconnections_flag/cloudsql_instance_postgres_log_disconnections_flag_test.py new file mode 100644 index 0000000000..57a58a23d1 --- /dev/null +++ b/tests/providers/gcp/services/cloudsql/cloudsql_instance_postgres_log_disconnections_flag/cloudsql_instance_postgres_log_disconnections_flag_test.py @@ -0,0 +1,200 @@ +from unittest import mock + +from tests.providers.gcp.gcp_fixtures import ( + GCP_EU1_LOCATION, + GCP_PROJECT_ID, + set_mocked_gcp_provider, +) + + +class Test_cloudsql_instance_postgres_log_disconnections_flag: + def test_no_cloudsql_instances(self): + cloudsql_client = mock.MagicMock + + with mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=set_mocked_gcp_provider(), + ), mock.patch( + "prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_disconnections_flag.cloudsql_instance_postgres_log_disconnections_flag.cloudsql_client", + new=cloudsql_client, + ): + from prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_disconnections_flag.cloudsql_instance_postgres_log_disconnections_flag import ( + cloudsql_instance_postgres_log_disconnections_flag, + ) + + cloudsql_client.instances = [] + + check = cloudsql_instance_postgres_log_disconnections_flag() + result = check.execute() + assert len(result) == 0 + + def test_cloudsql_mysql_instance(self): + cloudsql_client = mock.MagicMock + + with mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=set_mocked_gcp_provider(), + ), mock.patch( + "prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_disconnections_flag.cloudsql_instance_postgres_log_disconnections_flag.cloudsql_client", + new=cloudsql_client, + ): + from prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_disconnections_flag.cloudsql_instance_postgres_log_disconnections_flag import ( + cloudsql_instance_postgres_log_disconnections_flag, + ) + from prowler.providers.gcp.services.cloudsql.cloudsql_service import ( + Instance, + ) + + cloudsql_client.instances = [ + Instance( + name="instance1", + version="MYSQL_5_7", + ip_addresses=[], + region=GCP_EU1_LOCATION, + public_ip=False, + ssl=False, + automated_backups=True, + authorized_networks=[], + flags=[], + project_id=GCP_PROJECT_ID, + ) + ] + + check = cloudsql_instance_postgres_log_disconnections_flag() + result = check.execute() + assert len(result) == 0 + + def test_cloudsql_instance_no_flags(self): + cloudsql_client = mock.MagicMock + + with mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=set_mocked_gcp_provider(), + ), mock.patch( + "prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_disconnections_flag.cloudsql_instance_postgres_log_disconnections_flag.cloudsql_client", + new=cloudsql_client, + ): + from prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_disconnections_flag.cloudsql_instance_postgres_log_disconnections_flag import ( + cloudsql_instance_postgres_log_disconnections_flag, + ) + from prowler.providers.gcp.services.cloudsql.cloudsql_service import ( + Instance, + ) + + cloudsql_client.instances = [ + Instance( + name="instance1", + version="POSTGRES_15", + ip_addresses=[], + region=GCP_EU1_LOCATION, + public_ip=False, + ssl=False, + automated_backups=True, + authorized_networks=[], + flags=[], + project_id=GCP_PROJECT_ID, + ) + ] + + check = cloudsql_instance_postgres_log_disconnections_flag() + result = check.execute() + assert len(result) == 1 + assert result[0].status == "FAIL" + assert ( + result[0].status_extended + == "PostgreSQL Instance instance1 does not have 'log_disconnections' flag set to 'on'." + ) + assert result[0].resource_id == "instance1" + assert result[0].resource_name == "instance1" + assert result[0].location == GCP_EU1_LOCATION + assert result[0].project_id == GCP_PROJECT_ID + + def test_cloudsql_instance_log_disconnections_flag_off(self): + cloudsql_client = mock.MagicMock + + with mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=set_mocked_gcp_provider(), + ), mock.patch( + "prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_disconnections_flag.cloudsql_instance_postgres_log_disconnections_flag.cloudsql_client", + new=cloudsql_client, + ): + from prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_disconnections_flag.cloudsql_instance_postgres_log_disconnections_flag import ( + cloudsql_instance_postgres_log_disconnections_flag, + ) + from prowler.providers.gcp.services.cloudsql.cloudsql_service import ( + Instance, + ) + + cloudsql_client.instances = [ + Instance( + name="instance1", + version="POSTGRES_15", + ip_addresses=[], + region=GCP_EU1_LOCATION, + public_ip=False, + ssl=False, + automated_backups=True, + authorized_networks=[], + flags=[{"name": "log_disconnections", "value": "off"}], + project_id=GCP_PROJECT_ID, + ) + ] + + check = cloudsql_instance_postgres_log_disconnections_flag() + result = check.execute() + assert len(result) == 1 + assert result[0].status == "FAIL" + assert ( + result[0].status_extended + == "PostgreSQL Instance instance1 does not have 'log_disconnections' flag set to 'on'." + ) + assert result[0].resource_id == "instance1" + assert result[0].resource_name == "instance1" + assert result[0].location == GCP_EU1_LOCATION + assert result[0].project_id == GCP_PROJECT_ID + + def test_cloudsql_instance_log_disconnections_flag_on(self): + cloudsql_client = mock.MagicMock + + with mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=set_mocked_gcp_provider(), + ), mock.patch( + "prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_disconnections_flag.cloudsql_instance_postgres_log_disconnections_flag.cloudsql_client", + new=cloudsql_client, + ): + from prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_disconnections_flag.cloudsql_instance_postgres_log_disconnections_flag import ( + cloudsql_instance_postgres_log_disconnections_flag, + ) + from prowler.providers.gcp.services.cloudsql.cloudsql_service import ( + Instance, + ) + + cloudsql_client.instances = [ + Instance( + name="instance1", + version="POSTGRES_15", + ip_addresses=[], + region=GCP_EU1_LOCATION, + public_ip=False, + ssl=False, + automated_backups=True, + authorized_networks=[], + flags=[{"name": "log_disconnections", "value": "on"}], + project_id=GCP_PROJECT_ID, + ) + ] + + check = cloudsql_instance_postgres_log_disconnections_flag() + result = check.execute() + assert len(result) == 1 + assert result[0].status == "PASS" + assert ( + result[0].status_extended + == "PostgreSQL Instance instance1 has 'log_disconnections' flag set to 'on'." + ) + assert result[0].resource_id == "instance1" + assert result[0].resource_name == "instance1" + assert result[0].location == GCP_EU1_LOCATION + assert result[0].project_id == GCP_PROJECT_ID diff --git a/tests/providers/gcp/services/cloudsql/cloudsql_instance_postgres_log_error_verbosity_flag/cloudsql_instance_postgres_log_error_verbosity_flag_test.py b/tests/providers/gcp/services/cloudsql/cloudsql_instance_postgres_log_error_verbosity_flag/cloudsql_instance_postgres_log_error_verbosity_flag_test.py new file mode 100644 index 0000000000..ee0f26420f --- /dev/null +++ b/tests/providers/gcp/services/cloudsql/cloudsql_instance_postgres_log_error_verbosity_flag/cloudsql_instance_postgres_log_error_verbosity_flag_test.py @@ -0,0 +1,200 @@ +from unittest import mock + +from tests.providers.gcp.gcp_fixtures import ( + GCP_EU1_LOCATION, + GCP_PROJECT_ID, + set_mocked_gcp_provider, +) + + +class Test_cloudsql_instance_postgres_log_error_verbosity_flag: + def test_no_cloudsql_instances(self): + cloudsql_client = mock.MagicMock + + with mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=set_mocked_gcp_provider(), + ), mock.patch( + "prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_error_verbosity_flag.cloudsql_instance_postgres_log_error_verbosity_flag.cloudsql_client", + new=cloudsql_client, + ): + from prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_error_verbosity_flag.cloudsql_instance_postgres_log_error_verbosity_flag import ( + cloudsql_instance_postgres_log_error_verbosity_flag, + ) + + cloudsql_client.instances = [] + + check = cloudsql_instance_postgres_log_error_verbosity_flag() + result = check.execute() + assert len(result) == 0 + + def test_cloudsql_mysql_instance(self): + cloudsql_client = mock.MagicMock + + with mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=set_mocked_gcp_provider(), + ), mock.patch( + "prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_error_verbosity_flag.cloudsql_instance_postgres_log_error_verbosity_flag.cloudsql_client", + new=cloudsql_client, + ): + from prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_error_verbosity_flag.cloudsql_instance_postgres_log_error_verbosity_flag import ( + cloudsql_instance_postgres_log_error_verbosity_flag, + ) + from prowler.providers.gcp.services.cloudsql.cloudsql_service import ( + Instance, + ) + + cloudsql_client.instances = [ + Instance( + name="instance1", + version="MYSQL_5_7", + ip_addresses=[], + region=GCP_EU1_LOCATION, + public_ip=False, + ssl=False, + automated_backups=True, + authorized_networks=[], + flags=[], + project_id=GCP_PROJECT_ID, + ) + ] + + check = cloudsql_instance_postgres_log_error_verbosity_flag() + result = check.execute() + assert len(result) == 0 + + def test_cloudsql_instance_no_flags(self): + cloudsql_client = mock.MagicMock + + with mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=set_mocked_gcp_provider(), + ), mock.patch( + "prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_error_verbosity_flag.cloudsql_instance_postgres_log_error_verbosity_flag.cloudsql_client", + new=cloudsql_client, + ): + from prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_error_verbosity_flag.cloudsql_instance_postgres_log_error_verbosity_flag import ( + cloudsql_instance_postgres_log_error_verbosity_flag, + ) + from prowler.providers.gcp.services.cloudsql.cloudsql_service import ( + Instance, + ) + + cloudsql_client.instances = [ + Instance( + name="instance1", + version="POSTGRES_15", + ip_addresses=[], + region=GCP_EU1_LOCATION, + public_ip=False, + ssl=False, + automated_backups=True, + authorized_networks=[], + flags=[], + project_id=GCP_PROJECT_ID, + ) + ] + + check = cloudsql_instance_postgres_log_error_verbosity_flag() + result = check.execute() + assert len(result) == 1 + assert result[0].status == "FAIL" + assert ( + result[0].status_extended + == "PostgreSQL Instance instance1 does not have 'log_error_verbosity' flag set to 'default'." + ) + assert result[0].resource_id == "instance1" + assert result[0].resource_name == "instance1" + assert result[0].location == GCP_EU1_LOCATION + assert result[0].project_id == GCP_PROJECT_ID + + def test_cloudsql_instance_log_error_verbosity_flag_off(self): + cloudsql_client = mock.MagicMock + + with mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=set_mocked_gcp_provider(), + ), mock.patch( + "prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_error_verbosity_flag.cloudsql_instance_postgres_log_error_verbosity_flag.cloudsql_client", + new=cloudsql_client, + ): + from prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_error_verbosity_flag.cloudsql_instance_postgres_log_error_verbosity_flag import ( + cloudsql_instance_postgres_log_error_verbosity_flag, + ) + from prowler.providers.gcp.services.cloudsql.cloudsql_service import ( + Instance, + ) + + cloudsql_client.instances = [ + Instance( + name="instance1", + version="POSTGRES_15", + ip_addresses=[], + region=GCP_EU1_LOCATION, + public_ip=False, + ssl=False, + automated_backups=True, + authorized_networks=[], + flags=[{"name": "log_error_verbosity", "value": "off"}], + project_id=GCP_PROJECT_ID, + ) + ] + + check = cloudsql_instance_postgres_log_error_verbosity_flag() + result = check.execute() + assert len(result) == 1 + assert result[0].status == "FAIL" + assert ( + result[0].status_extended + == "PostgreSQL Instance instance1 does not have 'log_error_verbosity' flag set to 'default'." + ) + assert result[0].resource_id == "instance1" + assert result[0].resource_name == "instance1" + assert result[0].location == GCP_EU1_LOCATION + assert result[0].project_id == GCP_PROJECT_ID + + def test_cloudsql_instance_log_error_verbosity_flag_on(self): + cloudsql_client = mock.MagicMock + + with mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=set_mocked_gcp_provider(), + ), mock.patch( + "prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_error_verbosity_flag.cloudsql_instance_postgres_log_error_verbosity_flag.cloudsql_client", + new=cloudsql_client, + ): + from prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_error_verbosity_flag.cloudsql_instance_postgres_log_error_verbosity_flag import ( + cloudsql_instance_postgres_log_error_verbosity_flag, + ) + from prowler.providers.gcp.services.cloudsql.cloudsql_service import ( + Instance, + ) + + cloudsql_client.instances = [ + Instance( + name="instance1", + version="POSTGRES_15", + ip_addresses=[], + region=GCP_EU1_LOCATION, + public_ip=False, + ssl=False, + automated_backups=True, + authorized_networks=[], + flags=[{"name": "log_error_verbosity", "value": "default"}], + project_id=GCP_PROJECT_ID, + ) + ] + + check = cloudsql_instance_postgres_log_error_verbosity_flag() + result = check.execute() + assert len(result) == 1 + assert result[0].status == "PASS" + assert ( + result[0].status_extended + == "PostgreSQL Instance instance1 has 'log_error_verbosity' flag set to 'default'." + ) + assert result[0].resource_id == "instance1" + assert result[0].resource_name == "instance1" + assert result[0].location == GCP_EU1_LOCATION + assert result[0].project_id == GCP_PROJECT_ID diff --git a/tests/providers/gcp/services/cloudsql/cloudsql_instance_postgres_log_min_duration_statement_flag/cloudsql_instance_postgres_log_min_duration_statement_flag_test.py b/tests/providers/gcp/services/cloudsql/cloudsql_instance_postgres_log_min_duration_statement_flag/cloudsql_instance_postgres_log_min_duration_statement_flag_test.py new file mode 100644 index 0000000000..fc7f66b024 --- /dev/null +++ b/tests/providers/gcp/services/cloudsql/cloudsql_instance_postgres_log_min_duration_statement_flag/cloudsql_instance_postgres_log_min_duration_statement_flag_test.py @@ -0,0 +1,200 @@ +from unittest import mock + +from tests.providers.gcp.gcp_fixtures import ( + GCP_EU1_LOCATION, + GCP_PROJECT_ID, + set_mocked_gcp_provider, +) + + +class Test_cloudsql_instance_postgres_log_min_duration_statement_flag: + def test_no_cloudsql_instances(self): + cloudsql_client = mock.MagicMock + + with mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=set_mocked_gcp_provider(), + ), mock.patch( + "prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_min_duration_statement_flag.cloudsql_instance_postgres_log_min_duration_statement_flag.cloudsql_client", + new=cloudsql_client, + ): + from prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_min_duration_statement_flag.cloudsql_instance_postgres_log_min_duration_statement_flag import ( + cloudsql_instance_postgres_log_min_duration_statement_flag, + ) + + cloudsql_client.instances = [] + + check = cloudsql_instance_postgres_log_min_duration_statement_flag() + result = check.execute() + assert len(result) == 0 + + def test_cloudsql_mysql_instance(self): + cloudsql_client = mock.MagicMock + + with mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=set_mocked_gcp_provider(), + ), mock.patch( + "prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_min_duration_statement_flag.cloudsql_instance_postgres_log_min_duration_statement_flag.cloudsql_client", + new=cloudsql_client, + ): + from prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_min_duration_statement_flag.cloudsql_instance_postgres_log_min_duration_statement_flag import ( + cloudsql_instance_postgres_log_min_duration_statement_flag, + ) + from prowler.providers.gcp.services.cloudsql.cloudsql_service import ( + Instance, + ) + + cloudsql_client.instances = [ + Instance( + name="instance1", + version="MYSQL_5_7", + ip_addresses=[], + region=GCP_EU1_LOCATION, + public_ip=False, + ssl=False, + automated_backups=True, + authorized_networks=[], + flags=[], + project_id=GCP_PROJECT_ID, + ) + ] + + check = cloudsql_instance_postgres_log_min_duration_statement_flag() + result = check.execute() + assert len(result) == 0 + + def test_cloudsql_instance_no_flags(self): + cloudsql_client = mock.MagicMock + + with mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=set_mocked_gcp_provider(), + ), mock.patch( + "prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_min_duration_statement_flag.cloudsql_instance_postgres_log_min_duration_statement_flag.cloudsql_client", + new=cloudsql_client, + ): + from prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_min_duration_statement_flag.cloudsql_instance_postgres_log_min_duration_statement_flag import ( + cloudsql_instance_postgres_log_min_duration_statement_flag, + ) + from prowler.providers.gcp.services.cloudsql.cloudsql_service import ( + Instance, + ) + + cloudsql_client.instances = [ + Instance( + name="instance1", + version="POSTGRES_15", + ip_addresses=[], + region=GCP_EU1_LOCATION, + public_ip=False, + ssl=False, + automated_backups=True, + authorized_networks=[], + flags=[], + project_id=GCP_PROJECT_ID, + ) + ] + + check = cloudsql_instance_postgres_log_min_duration_statement_flag() + result = check.execute() + assert len(result) == 1 + assert result[0].status == "FAIL" + assert ( + result[0].status_extended + == "PostgreSQL Instance instance1 does not have 'log_min_duration_statement' flag set to '-1'." + ) + assert result[0].resource_id == "instance1" + assert result[0].resource_name == "instance1" + assert result[0].location == GCP_EU1_LOCATION + assert result[0].project_id == GCP_PROJECT_ID + + def test_cloudsql_instance_log_min_duration_statement_flag_off(self): + cloudsql_client = mock.MagicMock + + with mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=set_mocked_gcp_provider(), + ), mock.patch( + "prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_min_duration_statement_flag.cloudsql_instance_postgres_log_min_duration_statement_flag.cloudsql_client", + new=cloudsql_client, + ): + from prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_min_duration_statement_flag.cloudsql_instance_postgres_log_min_duration_statement_flag import ( + cloudsql_instance_postgres_log_min_duration_statement_flag, + ) + from prowler.providers.gcp.services.cloudsql.cloudsql_service import ( + Instance, + ) + + cloudsql_client.instances = [ + Instance( + name="instance1", + version="POSTGRES_15", + ip_addresses=[], + region=GCP_EU1_LOCATION, + public_ip=False, + ssl=False, + automated_backups=True, + authorized_networks=[], + flags=[{"name": "log_min_duration_statement", "value": "0"}], + project_id=GCP_PROJECT_ID, + ) + ] + + check = cloudsql_instance_postgres_log_min_duration_statement_flag() + result = check.execute() + assert len(result) == 1 + assert result[0].status == "FAIL" + assert ( + result[0].status_extended + == "PostgreSQL Instance instance1 does not have 'log_min_duration_statement' flag set to '-1'." + ) + assert result[0].resource_id == "instance1" + assert result[0].resource_name == "instance1" + assert result[0].location == GCP_EU1_LOCATION + assert result[0].project_id == GCP_PROJECT_ID + + def test_cloudsql_instance_log_min_duration_statement_flag_on(self): + cloudsql_client = mock.MagicMock + + with mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=set_mocked_gcp_provider(), + ), mock.patch( + "prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_min_duration_statement_flag.cloudsql_instance_postgres_log_min_duration_statement_flag.cloudsql_client", + new=cloudsql_client, + ): + from prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_min_duration_statement_flag.cloudsql_instance_postgres_log_min_duration_statement_flag import ( + cloudsql_instance_postgres_log_min_duration_statement_flag, + ) + from prowler.providers.gcp.services.cloudsql.cloudsql_service import ( + Instance, + ) + + cloudsql_client.instances = [ + Instance( + name="instance1", + version="POSTGRES_15", + ip_addresses=[], + region=GCP_EU1_LOCATION, + public_ip=False, + ssl=False, + automated_backups=True, + authorized_networks=[], + flags=[{"name": "log_min_duration_statement", "value": "-1"}], + project_id=GCP_PROJECT_ID, + ) + ] + + check = cloudsql_instance_postgres_log_min_duration_statement_flag() + result = check.execute() + assert len(result) == 1 + assert result[0].status == "PASS" + assert ( + result[0].status_extended + == "PostgreSQL Instance instance1 has 'log_min_duration_statement' flag set to '-1'." + ) + assert result[0].resource_id == "instance1" + assert result[0].resource_name == "instance1" + assert result[0].location == GCP_EU1_LOCATION + assert result[0].project_id == GCP_PROJECT_ID diff --git a/tests/providers/gcp/services/cloudsql/cloudsql_instance_postgres_log_min_error_statement_flag/cloudsql_instance_postgres_log_min_error_statement_flag_test.py b/tests/providers/gcp/services/cloudsql/cloudsql_instance_postgres_log_min_error_statement_flag/cloudsql_instance_postgres_log_min_error_statement_flag_test.py new file mode 100644 index 0000000000..08741bb348 --- /dev/null +++ b/tests/providers/gcp/services/cloudsql/cloudsql_instance_postgres_log_min_error_statement_flag/cloudsql_instance_postgres_log_min_error_statement_flag_test.py @@ -0,0 +1,200 @@ +from unittest import mock + +from tests.providers.gcp.gcp_fixtures import ( + GCP_EU1_LOCATION, + GCP_PROJECT_ID, + set_mocked_gcp_provider, +) + + +class Test_cloudsql_instance_postgres_log_min_error_statement_flag: + def test_no_cloudsql_instances(self): + cloudsql_client = mock.MagicMock + + with mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=set_mocked_gcp_provider(), + ), mock.patch( + "prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_min_error_statement_flag.cloudsql_instance_postgres_log_min_error_statement_flag.cloudsql_client", + new=cloudsql_client, + ): + from prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_min_error_statement_flag.cloudsql_instance_postgres_log_min_error_statement_flag import ( + cloudsql_instance_postgres_log_min_error_statement_flag, + ) + + cloudsql_client.instances = [] + + check = cloudsql_instance_postgres_log_min_error_statement_flag() + result = check.execute() + assert len(result) == 0 + + def test_cloudsql_mysql_instance(self): + cloudsql_client = mock.MagicMock + + with mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=set_mocked_gcp_provider(), + ), mock.patch( + "prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_min_error_statement_flag.cloudsql_instance_postgres_log_min_error_statement_flag.cloudsql_client", + new=cloudsql_client, + ): + from prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_min_error_statement_flag.cloudsql_instance_postgres_log_min_error_statement_flag import ( + cloudsql_instance_postgres_log_min_error_statement_flag, + ) + from prowler.providers.gcp.services.cloudsql.cloudsql_service import ( + Instance, + ) + + cloudsql_client.instances = [ + Instance( + name="instance1", + version="MYSQL_5_7", + ip_addresses=[], + region=GCP_EU1_LOCATION, + public_ip=False, + ssl=False, + automated_backups=True, + authorized_networks=[], + flags=[], + project_id=GCP_PROJECT_ID, + ) + ] + + check = cloudsql_instance_postgres_log_min_error_statement_flag() + result = check.execute() + assert len(result) == 0 + + def test_cloudsql_instance_no_flags(self): + cloudsql_client = mock.MagicMock + + with mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=set_mocked_gcp_provider(), + ), mock.patch( + "prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_min_error_statement_flag.cloudsql_instance_postgres_log_min_error_statement_flag.cloudsql_client", + new=cloudsql_client, + ): + from prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_min_error_statement_flag.cloudsql_instance_postgres_log_min_error_statement_flag import ( + cloudsql_instance_postgres_log_min_error_statement_flag, + ) + from prowler.providers.gcp.services.cloudsql.cloudsql_service import ( + Instance, + ) + + cloudsql_client.instances = [ + Instance( + name="instance1", + version="POSTGRES_15", + ip_addresses=[], + region=GCP_EU1_LOCATION, + public_ip=False, + ssl=False, + automated_backups=True, + authorized_networks=[], + flags=[], + project_id=GCP_PROJECT_ID, + ) + ] + + check = cloudsql_instance_postgres_log_min_error_statement_flag() + result = check.execute() + assert len(result) == 1 + assert result[0].status == "FAIL" + assert ( + result[0].status_extended + == "PostgreSQL Instance instance1 does not have 'log_min_error_statement' flag set minimum to 'error'." + ) + assert result[0].resource_id == "instance1" + assert result[0].resource_name == "instance1" + assert result[0].location == GCP_EU1_LOCATION + assert result[0].project_id == GCP_PROJECT_ID + + def test_cloudsql_instance_log_min_error_statement_flag_off(self): + cloudsql_client = mock.MagicMock + + with mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=set_mocked_gcp_provider(), + ), mock.patch( + "prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_min_error_statement_flag.cloudsql_instance_postgres_log_min_error_statement_flag.cloudsql_client", + new=cloudsql_client, + ): + from prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_min_error_statement_flag.cloudsql_instance_postgres_log_min_error_statement_flag import ( + cloudsql_instance_postgres_log_min_error_statement_flag, + ) + from prowler.providers.gcp.services.cloudsql.cloudsql_service import ( + Instance, + ) + + cloudsql_client.instances = [ + Instance( + name="instance1", + version="POSTGRES_15", + ip_addresses=[], + region=GCP_EU1_LOCATION, + public_ip=False, + ssl=False, + automated_backups=True, + authorized_networks=[], + flags=[{"name": "log_min_error_statement", "value": "warning"}], + project_id=GCP_PROJECT_ID, + ) + ] + + check = cloudsql_instance_postgres_log_min_error_statement_flag() + result = check.execute() + assert len(result) == 1 + assert result[0].status == "FAIL" + assert ( + result[0].status_extended + == "PostgreSQL Instance instance1 does not have 'log_min_error_statement' flag set minimum to 'error'." + ) + assert result[0].resource_id == "instance1" + assert result[0].resource_name == "instance1" + assert result[0].location == GCP_EU1_LOCATION + assert result[0].project_id == GCP_PROJECT_ID + + def test_cloudsql_instance_log_min_error_statement_flag_on(self): + cloudsql_client = mock.MagicMock + + with mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=set_mocked_gcp_provider(), + ), mock.patch( + "prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_min_error_statement_flag.cloudsql_instance_postgres_log_min_error_statement_flag.cloudsql_client", + new=cloudsql_client, + ): + from prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_min_error_statement_flag.cloudsql_instance_postgres_log_min_error_statement_flag import ( + cloudsql_instance_postgres_log_min_error_statement_flag, + ) + from prowler.providers.gcp.services.cloudsql.cloudsql_service import ( + Instance, + ) + + cloudsql_client.instances = [ + Instance( + name="instance1", + version="POSTGRES_15", + ip_addresses=[], + region=GCP_EU1_LOCATION, + public_ip=False, + ssl=False, + automated_backups=True, + authorized_networks=[], + flags=[{"name": "log_min_error_statement", "value": "error"}], + project_id=GCP_PROJECT_ID, + ) + ] + + check = cloudsql_instance_postgres_log_min_error_statement_flag() + result = check.execute() + assert len(result) == 1 + assert result[0].status == "PASS" + assert ( + result[0].status_extended + == "PostgreSQL Instance instance1 has 'log_min_error_statement' flag set minimum to 'error'." + ) + assert result[0].resource_id == "instance1" + assert result[0].resource_name == "instance1" + assert result[0].location == GCP_EU1_LOCATION + assert result[0].project_id == GCP_PROJECT_ID diff --git a/tests/providers/gcp/services/cloudsql/cloudsql_instance_postgres_log_min_messages_flag/cloudsql_instance_postgres_log_min_messages_flag_test.py b/tests/providers/gcp/services/cloudsql/cloudsql_instance_postgres_log_min_messages_flag/cloudsql_instance_postgres_log_min_messages_flag_test.py new file mode 100644 index 0000000000..95dd3c0802 --- /dev/null +++ b/tests/providers/gcp/services/cloudsql/cloudsql_instance_postgres_log_min_messages_flag/cloudsql_instance_postgres_log_min_messages_flag_test.py @@ -0,0 +1,200 @@ +from unittest import mock + +from tests.providers.gcp.gcp_fixtures import ( + GCP_EU1_LOCATION, + GCP_PROJECT_ID, + set_mocked_gcp_provider, +) + + +class Test_cloudsql_instance_postgres_log_min_messages_flag: + def test_no_cloudsql_instances(self): + cloudsql_client = mock.MagicMock + + with mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=set_mocked_gcp_provider(), + ), mock.patch( + "prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_min_messages_flag.cloudsql_instance_postgres_log_min_messages_flag.cloudsql_client", + new=cloudsql_client, + ): + from prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_min_messages_flag.cloudsql_instance_postgres_log_min_messages_flag import ( + cloudsql_instance_postgres_log_min_messages_flag, + ) + + cloudsql_client.instances = [] + + check = cloudsql_instance_postgres_log_min_messages_flag() + result = check.execute() + assert len(result) == 0 + + def test_cloudsql_mysql_instance(self): + cloudsql_client = mock.MagicMock + + with mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=set_mocked_gcp_provider(), + ), mock.patch( + "prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_min_messages_flag.cloudsql_instance_postgres_log_min_messages_flag.cloudsql_client", + new=cloudsql_client, + ): + from prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_min_messages_flag.cloudsql_instance_postgres_log_min_messages_flag import ( + cloudsql_instance_postgres_log_min_messages_flag, + ) + from prowler.providers.gcp.services.cloudsql.cloudsql_service import ( + Instance, + ) + + cloudsql_client.instances = [ + Instance( + name="instance1", + version="MYSQL_5_7", + ip_addresses=[], + region=GCP_EU1_LOCATION, + public_ip=False, + ssl=False, + automated_backups=True, + authorized_networks=[], + flags=[], + project_id=GCP_PROJECT_ID, + ) + ] + + check = cloudsql_instance_postgres_log_min_messages_flag() + result = check.execute() + assert len(result) == 0 + + def test_cloudsql_instance_no_flags(self): + cloudsql_client = mock.MagicMock + + with mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=set_mocked_gcp_provider(), + ), mock.patch( + "prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_min_messages_flag.cloudsql_instance_postgres_log_min_messages_flag.cloudsql_client", + new=cloudsql_client, + ): + from prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_min_messages_flag.cloudsql_instance_postgres_log_min_messages_flag import ( + cloudsql_instance_postgres_log_min_messages_flag, + ) + from prowler.providers.gcp.services.cloudsql.cloudsql_service import ( + Instance, + ) + + cloudsql_client.instances = [ + Instance( + name="instance1", + version="POSTGRES_15", + ip_addresses=[], + region=GCP_EU1_LOCATION, + public_ip=False, + ssl=False, + automated_backups=True, + authorized_networks=[], + flags=[], + project_id=GCP_PROJECT_ID, + ) + ] + + check = cloudsql_instance_postgres_log_min_messages_flag() + result = check.execute() + assert len(result) == 1 + assert result[0].status == "FAIL" + assert ( + result[0].status_extended + == "PostgreSQL Instance instance1 does not have 'log_min_messages' flag set minimum to 'error'." + ) + assert result[0].resource_id == "instance1" + assert result[0].resource_name == "instance1" + assert result[0].location == GCP_EU1_LOCATION + assert result[0].project_id == GCP_PROJECT_ID + + def test_cloudsql_instance_log_min_messages_flag_off(self): + cloudsql_client = mock.MagicMock + + with mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=set_mocked_gcp_provider(), + ), mock.patch( + "prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_min_messages_flag.cloudsql_instance_postgres_log_min_messages_flag.cloudsql_client", + new=cloudsql_client, + ): + from prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_min_messages_flag.cloudsql_instance_postgres_log_min_messages_flag import ( + cloudsql_instance_postgres_log_min_messages_flag, + ) + from prowler.providers.gcp.services.cloudsql.cloudsql_service import ( + Instance, + ) + + cloudsql_client.instances = [ + Instance( + name="instance1", + version="POSTGRES_15", + ip_addresses=[], + region=GCP_EU1_LOCATION, + public_ip=False, + ssl=False, + automated_backups=True, + authorized_networks=[], + flags=[{"name": "log_min_messages", "value": "debug"}], + project_id=GCP_PROJECT_ID, + ) + ] + + check = cloudsql_instance_postgres_log_min_messages_flag() + result = check.execute() + assert len(result) == 1 + assert result[0].status == "FAIL" + assert ( + result[0].status_extended + == "PostgreSQL Instance instance1 does not have 'log_min_messages' flag set minimum to 'error'." + ) + assert result[0].resource_id == "instance1" + assert result[0].resource_name == "instance1" + assert result[0].location == GCP_EU1_LOCATION + assert result[0].project_id == GCP_PROJECT_ID + + def test_cloudsql_instance_log_min_messages_flag_on(self): + cloudsql_client = mock.MagicMock + + with mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=set_mocked_gcp_provider(), + ), mock.patch( + "prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_min_messages_flag.cloudsql_instance_postgres_log_min_messages_flag.cloudsql_client", + new=cloudsql_client, + ): + from prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_min_messages_flag.cloudsql_instance_postgres_log_min_messages_flag import ( + cloudsql_instance_postgres_log_min_messages_flag, + ) + from prowler.providers.gcp.services.cloudsql.cloudsql_service import ( + Instance, + ) + + cloudsql_client.instances = [ + Instance( + name="instance1", + version="POSTGRES_15", + ip_addresses=[], + region=GCP_EU1_LOCATION, + public_ip=False, + ssl=False, + automated_backups=True, + authorized_networks=[], + flags=[{"name": "log_min_messages", "value": "error"}], + project_id=GCP_PROJECT_ID, + ) + ] + + check = cloudsql_instance_postgres_log_min_messages_flag() + result = check.execute() + assert len(result) == 1 + assert result[0].status == "PASS" + assert ( + result[0].status_extended + == "PostgreSQL Instance instance1 has 'log_min_messages' flag set minimum to 'error'." + ) + assert result[0].resource_id == "instance1" + assert result[0].resource_name == "instance1" + assert result[0].location == GCP_EU1_LOCATION + assert result[0].project_id == GCP_PROJECT_ID diff --git a/tests/providers/gcp/services/cloudsql/cloudsql_instance_postgres_log_statement_flag/cloudsql_instance_postgres_log_statement_flag_test.py b/tests/providers/gcp/services/cloudsql/cloudsql_instance_postgres_log_statement_flag/cloudsql_instance_postgres_log_statement_flag_test.py new file mode 100644 index 0000000000..67ad6e3a2b --- /dev/null +++ b/tests/providers/gcp/services/cloudsql/cloudsql_instance_postgres_log_statement_flag/cloudsql_instance_postgres_log_statement_flag_test.py @@ -0,0 +1,200 @@ +from unittest import mock + +from tests.providers.gcp.gcp_fixtures import ( + GCP_EU1_LOCATION, + GCP_PROJECT_ID, + set_mocked_gcp_provider, +) + + +class Test_cloudsql_instance_postgres_log_statement_flag: + def test_no_cloudsql_instances(self): + cloudsql_client = mock.MagicMock + + with mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=set_mocked_gcp_provider(), + ), mock.patch( + "prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_statement_flag.cloudsql_instance_postgres_log_statement_flag.cloudsql_client", + new=cloudsql_client, + ): + from prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_statement_flag.cloudsql_instance_postgres_log_statement_flag import ( + cloudsql_instance_postgres_log_statement_flag, + ) + + cloudsql_client.instances = [] + + check = cloudsql_instance_postgres_log_statement_flag() + result = check.execute() + assert len(result) == 0 + + def test_cloudsql_mysql_instance(self): + cloudsql_client = mock.MagicMock + + with mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=set_mocked_gcp_provider(), + ), mock.patch( + "prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_statement_flag.cloudsql_instance_postgres_log_statement_flag.cloudsql_client", + new=cloudsql_client, + ): + from prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_statement_flag.cloudsql_instance_postgres_log_statement_flag import ( + cloudsql_instance_postgres_log_statement_flag, + ) + from prowler.providers.gcp.services.cloudsql.cloudsql_service import ( + Instance, + ) + + cloudsql_client.instances = [ + Instance( + name="instance1", + version="MYSQL_5_7", + ip_addresses=[], + region=GCP_EU1_LOCATION, + public_ip=False, + ssl=False, + automated_backups=True, + authorized_networks=[], + flags=[], + project_id=GCP_PROJECT_ID, + ) + ] + + check = cloudsql_instance_postgres_log_statement_flag() + result = check.execute() + assert len(result) == 0 + + def test_cloudsql_instance_no_flags(self): + cloudsql_client = mock.MagicMock + + with mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=set_mocked_gcp_provider(), + ), mock.patch( + "prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_statement_flag.cloudsql_instance_postgres_log_statement_flag.cloudsql_client", + new=cloudsql_client, + ): + from prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_statement_flag.cloudsql_instance_postgres_log_statement_flag import ( + cloudsql_instance_postgres_log_statement_flag, + ) + from prowler.providers.gcp.services.cloudsql.cloudsql_service import ( + Instance, + ) + + cloudsql_client.instances = [ + Instance( + name="instance1", + version="POSTGRES_15", + ip_addresses=[], + region=GCP_EU1_LOCATION, + public_ip=False, + ssl=False, + automated_backups=True, + authorized_networks=[], + flags=[], + project_id=GCP_PROJECT_ID, + ) + ] + + check = cloudsql_instance_postgres_log_statement_flag() + result = check.execute() + assert len(result) == 1 + assert result[0].status == "FAIL" + assert ( + result[0].status_extended + == "PostgreSQL Instance instance1 does not have 'log_statement' flag set to 'ddl'." + ) + assert result[0].resource_id == "instance1" + assert result[0].resource_name == "instance1" + assert result[0].location == GCP_EU1_LOCATION + assert result[0].project_id == GCP_PROJECT_ID + + def test_cloudsql_instance_log_statement_flag_off(self): + cloudsql_client = mock.MagicMock + + with mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=set_mocked_gcp_provider(), + ), mock.patch( + "prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_statement_flag.cloudsql_instance_postgres_log_statement_flag.cloudsql_client", + new=cloudsql_client, + ): + from prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_statement_flag.cloudsql_instance_postgres_log_statement_flag import ( + cloudsql_instance_postgres_log_statement_flag, + ) + from prowler.providers.gcp.services.cloudsql.cloudsql_service import ( + Instance, + ) + + cloudsql_client.instances = [ + Instance( + name="instance1", + version="POSTGRES_15", + ip_addresses=[], + region=GCP_EU1_LOCATION, + public_ip=False, + ssl=False, + automated_backups=True, + authorized_networks=[], + flags=[{"name": "log_statement", "value": "all"}], + project_id=GCP_PROJECT_ID, + ) + ] + + check = cloudsql_instance_postgres_log_statement_flag() + result = check.execute() + assert len(result) == 1 + assert result[0].status == "FAIL" + assert ( + result[0].status_extended + == "PostgreSQL Instance instance1 does not have 'log_statement' flag set to 'ddl'." + ) + assert result[0].resource_id == "instance1" + assert result[0].resource_name == "instance1" + assert result[0].location == GCP_EU1_LOCATION + assert result[0].project_id == GCP_PROJECT_ID + + def test_cloudsql_instance_log_statement_flag_on(self): + cloudsql_client = mock.MagicMock + + with mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=set_mocked_gcp_provider(), + ), mock.patch( + "prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_statement_flag.cloudsql_instance_postgres_log_statement_flag.cloudsql_client", + new=cloudsql_client, + ): + from prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_statement_flag.cloudsql_instance_postgres_log_statement_flag import ( + cloudsql_instance_postgres_log_statement_flag, + ) + from prowler.providers.gcp.services.cloudsql.cloudsql_service import ( + Instance, + ) + + cloudsql_client.instances = [ + Instance( + name="instance1", + version="POSTGRES_15", + ip_addresses=[], + region=GCP_EU1_LOCATION, + public_ip=False, + ssl=False, + automated_backups=True, + authorized_networks=[], + flags=[{"name": "log_statement", "value": "ddl"}], + project_id=GCP_PROJECT_ID, + ) + ] + + check = cloudsql_instance_postgres_log_statement_flag() + result = check.execute() + assert len(result) == 1 + assert result[0].status == "PASS" + assert ( + result[0].status_extended + == "PostgreSQL Instance instance1 has 'log_statement' flag set to 'ddl'." + ) + assert result[0].resource_id == "instance1" + assert result[0].resource_name == "instance1" + assert result[0].location == GCP_EU1_LOCATION + assert result[0].project_id == GCP_PROJECT_ID