From 28f6f0abcc49205dab04271819d3d99de3042d0c Mon Sep 17 00:00:00 2001 From: Daniel Barranquero <74871504+danibarranqueroo@users.noreply.github.com> Date: Fri, 19 Jul 2024 14:37:22 +0200 Subject: [PATCH] test(cloudstorage): Add remaining GCP tests for CloudStorage checks (#4464) --- ...e_bucket_log_retention_policy_lock_test.py | 257 ++++++++++++++++++ .../cloudstorage_bucket_public_access_test.py | 122 +++++++++ ...bucket_uniform_bucket_level_access_test.py | 122 +++++++++ 3 files changed, 501 insertions(+) create mode 100644 tests/providers/gcp/services/cloudstorage/cloudstorage_bucket_log_retention_policy_lock/cloudstorage_bucket_log_retention_policy_lock_test.py create mode 100644 tests/providers/gcp/services/cloudstorage/cloudstorage_bucket_public_access/cloudstorage_bucket_public_access_test.py create mode 100644 tests/providers/gcp/services/cloudstorage/cloudstorage_bucket_uniform_bucket_level_access/cloudstorage_bucket_uniform_bucket_level_access_test.py diff --git a/tests/providers/gcp/services/cloudstorage/cloudstorage_bucket_log_retention_policy_lock/cloudstorage_bucket_log_retention_policy_lock_test.py b/tests/providers/gcp/services/cloudstorage/cloudstorage_bucket_log_retention_policy_lock/cloudstorage_bucket_log_retention_policy_lock_test.py new file mode 100644 index 0000000000..6c35fdfc2e --- /dev/null +++ b/tests/providers/gcp/services/cloudstorage/cloudstorage_bucket_log_retention_policy_lock/cloudstorage_bucket_log_retention_policy_lock_test.py @@ -0,0 +1,257 @@ +from unittest import mock + +from tests.providers.gcp.gcp_fixtures import ( + GCP_PROJECT_ID, + GCP_US_CENTER1_LOCATION, + set_mocked_gcp_provider, +) + + +class TestCloudStorageBucketLogRetentionPolicyLock: + def test_bucket_with_retention_policy_and_lock(self): + cloudstorage_client = mock.MagicMock() + logging_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.cloudstorage.cloudstorage_bucket_log_retention_policy_lock.cloudstorage_bucket_log_retention_policy_lock.cloudstorage_client", + new=cloudstorage_client, + ), mock.patch( + "prowler.providers.gcp.services.cloudstorage.cloudstorage_bucket_log_retention_policy_lock.cloudstorage_bucket_log_retention_policy_lock.logging_client", + new=logging_client, + ): + from prowler.providers.gcp.services.cloudstorage.cloudstorage_bucket_log_retention_policy_lock.cloudstorage_bucket_log_retention_policy_lock import ( + cloudstorage_bucket_log_retention_policy_lock, + ) + from prowler.providers.gcp.services.cloudstorage.cloudstorage_service import ( + Bucket, + ) + from prowler.providers.gcp.services.logging.logging_service import Sink + + cloudstorage_client.project_ids = [GCP_PROJECT_ID] + cloudstorage_client.region = GCP_US_CENTER1_LOCATION + + logging_client.sinks = [ + Sink( + name="sink1", + destination="storage.googleapis.com/example-bucket", + filter="all", + project_id=GCP_PROJECT_ID, + ) + ] + + cloudstorage_client.buckets = [ + Bucket( + name="example-bucket", + id="example-bucket", + region=GCP_US_CENTER1_LOCATION, + uniform_bucket_level_access=True, + public=True, + retention_policy={"isLocked": True}, + project_id=GCP_PROJECT_ID, + ) + ] + + check = cloudstorage_bucket_log_retention_policy_lock() + result = check.execute() + + assert len(result) == 1 + assert result[0].status == "PASS" + assert ( + result[0].status_extended + == f"Log Sink Bucket {cloudstorage_client.buckets[0].name} has a Retention Policy with Bucket Lock." + ) + assert result[0].resource_id == "example-bucket" + assert result[0].resource_name == "example-bucket" + assert result[0].location == GCP_US_CENTER1_LOCATION + assert result[0].project_id == GCP_PROJECT_ID + + def test_bucket_with_retention_policy_without_lock(self): + cloudstorage_client = mock.MagicMock() + logging_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.cloudstorage.cloudstorage_bucket_log_retention_policy_lock.cloudstorage_bucket_log_retention_policy_lock.cloudstorage_client", + new=cloudstorage_client, + ), mock.patch( + "prowler.providers.gcp.services.cloudstorage.cloudstorage_bucket_log_retention_policy_lock.cloudstorage_bucket_log_retention_policy_lock.logging_client", + new=logging_client, + ): + from prowler.providers.gcp.services.cloudstorage.cloudstorage_bucket_log_retention_policy_lock.cloudstorage_bucket_log_retention_policy_lock import ( + cloudstorage_bucket_log_retention_policy_lock, + ) + from prowler.providers.gcp.services.cloudstorage.cloudstorage_service import ( + Bucket, + ) + from prowler.providers.gcp.services.logging.logging_service import Sink + + cloudstorage_client.project_ids = [GCP_PROJECT_ID] + cloudstorage_client.region = GCP_US_CENTER1_LOCATION + + logging_client.sinks = [ + Sink( + name="sink1", + destination="storage.googleapis.com/example-bucket", + filter="all", + project_id=GCP_PROJECT_ID, + ) + ] + + cloudstorage_client.buckets = [ + Bucket( + name="example-bucket", + id="example-bucket", + region=GCP_US_CENTER1_LOCATION, + uniform_bucket_level_access=True, + public=True, + retention_policy={"isLocked": False}, + project_id=GCP_PROJECT_ID, + ) + ] + + check = cloudstorage_bucket_log_retention_policy_lock() + result = check.execute() + + assert len(result) == 1 + assert result[0].status == "FAIL" + assert ( + result[0].status_extended + == f"Log Sink Bucket {cloudstorage_client.buckets[0].name} has no Retention Policy but without Bucket Lock." + ) + assert result[0].resource_id == "example-bucket" + assert result[0].resource_name == "example-bucket" + assert result[0].location == GCP_US_CENTER1_LOCATION + assert result[0].project_id == GCP_PROJECT_ID + + def test_bucket_without_retention_policy(self): + cloudstorage_client = mock.MagicMock() + logging_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.cloudstorage.cloudstorage_bucket_log_retention_policy_lock.cloudstorage_bucket_log_retention_policy_lock.cloudstorage_client", + new=cloudstorage_client, + ), mock.patch( + "prowler.providers.gcp.services.cloudstorage.cloudstorage_bucket_log_retention_policy_lock.cloudstorage_bucket_log_retention_policy_lock.logging_client", + new=logging_client, + ): + from prowler.providers.gcp.services.cloudstorage.cloudstorage_bucket_log_retention_policy_lock.cloudstorage_bucket_log_retention_policy_lock import ( + cloudstorage_bucket_log_retention_policy_lock, + ) + from prowler.providers.gcp.services.cloudstorage.cloudstorage_service import ( + Bucket, + ) + from prowler.providers.gcp.services.logging.logging_service import Sink + + cloudstorage_client.project_ids = [GCP_PROJECT_ID] + cloudstorage_client.region = GCP_US_CENTER1_LOCATION + + logging_client.sinks = [ + Sink( + name="sink1", + destination="storage.googleapis.com/example-bucket", + filter="all", + project_id=GCP_PROJECT_ID, + ) + ] + + cloudstorage_client.buckets = [ + Bucket( + name="example-bucket", + id="example-bucket", + region=GCP_US_CENTER1_LOCATION, + uniform_bucket_level_access=True, + public=True, + retention_policy=None, + project_id=GCP_PROJECT_ID, + ) + ] + + check = cloudstorage_bucket_log_retention_policy_lock() + result = check.execute() + + assert len(result) == 1 + assert result[0].status == "FAIL" + assert ( + result[0].status_extended + == f"Log Sink Bucket {cloudstorage_client.buckets[0].name} has no Retention Policy." + ) + assert result[0].resource_id == "example-bucket" + assert result[0].resource_name == "example-bucket" + assert result[0].location == GCP_US_CENTER1_LOCATION + assert result[0].project_id == GCP_PROJECT_ID + + def test_no_buckets(self): + cloudstorage_client = mock.MagicMock() + logging_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.cloudstorage.cloudstorage_bucket_log_retention_policy_lock.cloudstorage_bucket_log_retention_policy_lock.cloudstorage_client", + new=cloudstorage_client, + ), mock.patch( + "prowler.providers.gcp.services.cloudstorage.cloudstorage_bucket_log_retention_policy_lock.cloudstorage_bucket_log_retention_policy_lock.logging_client", + new=logging_client, + ): + from prowler.providers.gcp.services.cloudstorage.cloudstorage_bucket_log_retention_policy_lock.cloudstorage_bucket_log_retention_policy_lock import ( + cloudstorage_bucket_log_retention_policy_lock, + ) + from prowler.providers.gcp.services.logging.logging_service import Sink + + cloudstorage_client.project_ids = [GCP_PROJECT_ID] + cloudstorage_client.region = GCP_US_CENTER1_LOCATION + + logging_client.sinks = [ + Sink( + name="sink1", + destination="storage.googleapis.com/example-bucket", + filter="all", + project_id=GCP_PROJECT_ID, + ) + ] + + cloudstorage_client.buckets = [] + + check = cloudstorage_bucket_log_retention_policy_lock() + result = check.execute() + + assert len(result) == 0 + + def test_no_buckets_no_sinks(self): + cloudstorage_client = mock.MagicMock() + logging_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.cloudstorage.cloudstorage_bucket_log_retention_policy_lock.cloudstorage_bucket_log_retention_policy_lock.cloudstorage_client", + new=cloudstorage_client, + ), mock.patch( + "prowler.providers.gcp.services.cloudstorage.cloudstorage_bucket_log_retention_policy_lock.cloudstorage_bucket_log_retention_policy_lock.logging_client", + new=logging_client, + ): + from prowler.providers.gcp.services.cloudstorage.cloudstorage_bucket_log_retention_policy_lock.cloudstorage_bucket_log_retention_policy_lock import ( + cloudstorage_bucket_log_retention_policy_lock, + ) + + cloudstorage_client.project_ids = [GCP_PROJECT_ID] + cloudstorage_client.region = GCP_US_CENTER1_LOCATION + + logging_client.sinks = [] + + cloudstorage_client.buckets = [] + + check = cloudstorage_bucket_log_retention_policy_lock() + result = check.execute() + + assert len(result) == 0 diff --git a/tests/providers/gcp/services/cloudstorage/cloudstorage_bucket_public_access/cloudstorage_bucket_public_access_test.py b/tests/providers/gcp/services/cloudstorage/cloudstorage_bucket_public_access/cloudstorage_bucket_public_access_test.py new file mode 100644 index 0000000000..1e78bebb28 --- /dev/null +++ b/tests/providers/gcp/services/cloudstorage/cloudstorage_bucket_public_access/cloudstorage_bucket_public_access_test.py @@ -0,0 +1,122 @@ +from unittest import mock + +from tests.providers.gcp.gcp_fixtures import ( + GCP_PROJECT_ID, + GCP_US_CENTER1_LOCATION, + set_mocked_gcp_provider, +) + + +class TestCloudStorageBucketPublicAccess: + def test_bucket_public_access(self): + cloudstorage_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.cloudstorage.cloudstorage_bucket_public_access.cloudstorage_bucket_public_access.cloudstorage_client", + new=cloudstorage_client, + ): + from prowler.providers.gcp.services.cloudstorage.cloudstorage_bucket_public_access.cloudstorage_bucket_public_access import ( + cloudstorage_bucket_public_access, + ) + from prowler.providers.gcp.services.cloudstorage.cloudstorage_service import ( + Bucket, + ) + + cloudstorage_client.project_ids = [GCP_PROJECT_ID] + cloudstorage_client.region = GCP_US_CENTER1_LOCATION + + cloudstorage_client.buckets = [ + Bucket( + name="example-bucket", + id="example-bucket", + region=GCP_US_CENTER1_LOCATION, + uniform_bucket_level_access=True, + public=True, + project_id=GCP_PROJECT_ID, + ) + ] + + check = cloudstorage_bucket_public_access() + result = check.execute() + + assert len(result) == 1 + assert result[0].status == "FAIL" + assert ( + result[0].status_extended + == f"Bucket {cloudstorage_client.buckets[0].name} is publicly accessible." + ) + assert result[0].resource_id == "example-bucket" + assert result[0].resource_name == "example-bucket" + assert result[0].location == GCP_US_CENTER1_LOCATION + assert result[0].project_id == GCP_PROJECT_ID + + def test_bucket_no_public_access(self): + cloudstorage_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.cloudstorage.cloudstorage_bucket_public_access.cloudstorage_bucket_public_access.cloudstorage_client", + new=cloudstorage_client, + ): + from prowler.providers.gcp.services.cloudstorage.cloudstorage_bucket_public_access.cloudstorage_bucket_public_access import ( + cloudstorage_bucket_public_access, + ) + from prowler.providers.gcp.services.cloudstorage.cloudstorage_service import ( + Bucket, + ) + + cloudstorage_client.project_ids = [GCP_PROJECT_ID] + cloudstorage_client.region = GCP_US_CENTER1_LOCATION + + cloudstorage_client.buckets = [ + Bucket( + name="example-bucket", + id="example-bucket", + region=GCP_US_CENTER1_LOCATION, + uniform_bucket_level_access=True, + public=False, + project_id=GCP_PROJECT_ID, + ) + ] + + check = cloudstorage_bucket_public_access() + result = check.execute() + + assert len(result) == 1 + assert result[0].status == "PASS" + assert ( + result[0].status_extended + == f"Bucket {cloudstorage_client.buckets[0].name} is not publicly accessible." + ) + assert result[0].resource_id == "example-bucket" + assert result[0].resource_name == "example-bucket" + assert result[0].location == GCP_US_CENTER1_LOCATION + assert result[0].project_id == GCP_PROJECT_ID + + def test_no_buckets(self): + cloudstorage_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.cloudstorage.cloudstorage_bucket_public_access.cloudstorage_bucket_public_access.cloudstorage_client", + new=cloudstorage_client, + ): + from prowler.providers.gcp.services.cloudstorage.cloudstorage_bucket_public_access.cloudstorage_bucket_public_access import ( + cloudstorage_bucket_public_access, + ) + + cloudstorage_client.project_ids = [GCP_PROJECT_ID] + cloudstorage_client.region = GCP_US_CENTER1_LOCATION + cloudstorage_client.buckets = [] + + check = cloudstorage_bucket_public_access() + result = check.execute() + + assert len(result) == 0 diff --git a/tests/providers/gcp/services/cloudstorage/cloudstorage_bucket_uniform_bucket_level_access/cloudstorage_bucket_uniform_bucket_level_access_test.py b/tests/providers/gcp/services/cloudstorage/cloudstorage_bucket_uniform_bucket_level_access/cloudstorage_bucket_uniform_bucket_level_access_test.py new file mode 100644 index 0000000000..7f7fe4e361 --- /dev/null +++ b/tests/providers/gcp/services/cloudstorage/cloudstorage_bucket_uniform_bucket_level_access/cloudstorage_bucket_uniform_bucket_level_access_test.py @@ -0,0 +1,122 @@ +from unittest import mock + +from tests.providers.gcp.gcp_fixtures import ( + GCP_PROJECT_ID, + GCP_US_CENTER1_LOCATION, + set_mocked_gcp_provider, +) + + +class TestCloudStorageBucketUniformBucketLevelAccess: + def test_bucket_with_uniform_bucket_level_access_enabled(self): + cloudstorage_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.cloudstorage.cloudstorage_bucket_uniform_bucket_level_access.cloudstorage_bucket_uniform_bucket_level_access.cloudstorage_client", + new=cloudstorage_client, + ): + from prowler.providers.gcp.services.cloudstorage.cloudstorage_bucket_uniform_bucket_level_access.cloudstorage_bucket_uniform_bucket_level_access import ( + cloudstorage_bucket_uniform_bucket_level_access, + ) + from prowler.providers.gcp.services.cloudstorage.cloudstorage_service import ( + Bucket, + ) + + cloudstorage_client.project_ids = [GCP_PROJECT_ID] + cloudstorage_client.region = GCP_US_CENTER1_LOCATION + + cloudstorage_client.buckets = [ + Bucket( + name="example-bucket", + id="example-bucket", + region=GCP_US_CENTER1_LOCATION, + uniform_bucket_level_access=True, + public=False, + project_id=GCP_PROJECT_ID, + ) + ] + + check = cloudstorage_bucket_uniform_bucket_level_access() + result = check.execute() + + assert len(result) == 1 + assert result[0].status == "PASS" + assert ( + result[0].status_extended + == f"Bucket {cloudstorage_client.buckets[0].name} has uniform Bucket Level Access enabled." + ) + assert result[0].resource_id == "example-bucket" + assert result[0].resource_name == "example-bucket" + assert result[0].location == GCP_US_CENTER1_LOCATION + assert result[0].project_id == GCP_PROJECT_ID + + def test_bucket_with_uniform_bucket_level_access_disabled(self): + cloudstorage_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.cloudstorage.cloudstorage_bucket_uniform_bucket_level_access.cloudstorage_bucket_uniform_bucket_level_access.cloudstorage_client", + new=cloudstorage_client, + ): + from prowler.providers.gcp.services.cloudstorage.cloudstorage_bucket_uniform_bucket_level_access.cloudstorage_bucket_uniform_bucket_level_access import ( + cloudstorage_bucket_uniform_bucket_level_access, + ) + from prowler.providers.gcp.services.cloudstorage.cloudstorage_service import ( + Bucket, + ) + + cloudstorage_client.project_ids = [GCP_PROJECT_ID] + cloudstorage_client.region = GCP_US_CENTER1_LOCATION + + cloudstorage_client.buckets = [ + Bucket( + name="example-bucket", + id="example-bucket", + region=GCP_US_CENTER1_LOCATION, + uniform_bucket_level_access=False, + public=False, + project_id=GCP_PROJECT_ID, + ) + ] + + check = cloudstorage_bucket_uniform_bucket_level_access() + result = check.execute() + + assert len(result) == 1 + assert result[0].status == "FAIL" + assert ( + result[0].status_extended + == f"Bucket {cloudstorage_client.buckets[0].name} has uniform Bucket Level Access disabled." + ) + assert result[0].resource_id == "example-bucket" + assert result[0].resource_name == "example-bucket" + assert result[0].location == GCP_US_CENTER1_LOCATION + assert result[0].project_id == GCP_PROJECT_ID + + def test_no_buckets(self): + cloudstorage_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.cloudstorage.cloudstorage_bucket_uniform_bucket_level_access.cloudstorage_bucket_uniform_bucket_level_access.cloudstorage_client", + new=cloudstorage_client, + ): + from prowler.providers.gcp.services.cloudstorage.cloudstorage_bucket_uniform_bucket_level_access.cloudstorage_bucket_uniform_bucket_level_access import ( + cloudstorage_bucket_uniform_bucket_level_access, + ) + + cloudstorage_client.project_ids = [GCP_PROJECT_ID] + cloudstorage_client.region = GCP_US_CENTER1_LOCATION + cloudstorage_client.buckets = [] + + check = cloudstorage_bucket_uniform_bucket_level_access() + result = check.execute() + + assert len(result) == 0