ben.carrasco/prowler
1
0
Fork 0
mirror of https://github.com/prowler-cloud/prowler.git synced 2026-07-24 21:11:53 +00:00
Code Issues Packages Projects Releases Wiki Activity

test(cloudstorage): Add remaining GCP tests for CloudStorage checks (#4464)

Browse Source
This commit is contained in:
Daniel Barranquero
2024-07-19 14:37:22 +02:00
committed by GitHub
parent 0933a04239
commit 28f6f0abcc
3 changed files with 501 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/providers/gcp/services/cloudstorage/cloudstorage_bucket_log_retention_policy_lock/cloudstorage_bucket_log_retention_policy_lock_test.py
+257
View File
@@ -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
tests/providers/gcp/services/cloudstorage/cloudstorage_bucket_public_access/cloudstorage_bucket_public_access_test.py
+122
View File
@@ -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
tests/providers/gcp/services/cloudstorage/cloudstorage_bucket_uniform_bucket_level_access/cloudstorage_bucket_uniform_bucket_level_access_test.py
+122
View File
@@ -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