ben.carrasco/prowler
1
0
Fork 0
mirror of https://github.com/prowler-cloud/prowler.git synced 2026-07-23 12:31:54 +00:00
Code Issues Packages Projects Releases Wiki Activity

test(logging): Add remaining tests for Logging checks (#4481)

Browse Source
This commit is contained in:
Rubén De la Torre Vico
2024-07-23 15:24:32 +02:00
committed by GitHub
parent 489830f01a
commit ca98584ded
9 changed files with 1603 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/providers/gcp/services/logging/logging_log_metric_filter_and_alert_for_audit_configuration_changes_enabled/logging_log_metric_filter_and_alert_for_audit_configuration_changes_enabled_test.py
+183
View File
@@ -0,0 +1,183 @@
from unittest.mock import MagicMock, patch
from tests.providers.gcp.gcp_fixtures import (
GCP_EU1_LOCATION,
GCP_PROJECT_ID,
set_mocked_gcp_provider,
)
class Test_logging_log_metric_filter_and_alert_for_audit_configuration_changes_enabled:
def test_no_projects(self):
logging_client = MagicMock()
monitoring_client = MagicMock()
with patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_gcp_provider(),
), patch(
"prowler.providers.gcp.services.logging.logging_log_metric_filter_and_alert_for_audit_configuration_changes_enabled.logging_log_metric_filter_and_alert_for_audit_configuration_changes_enabled.logging_client",
new=logging_client,
), patch(
"prowler.providers.gcp.services.logging.logging_log_metric_filter_and_alert_for_audit_configuration_changes_enabled.logging_log_metric_filter_and_alert_for_audit_configuration_changes_enabled.monitoring_client",
new=monitoring_client,
):
from prowler.providers.gcp.services.logging.logging_log_metric_filter_and_alert_for_audit_configuration_changes_enabled.logging_log_metric_filter_and_alert_for_audit_configuration_changes_enabled import (
logging_log_metric_filter_and_alert_for_audit_configuration_changes_enabled,
)
logging_client.metrics = []
logging_client.project_ids = []
monitoring_client.alert_policies = []
check = (
logging_log_metric_filter_and_alert_for_audit_configuration_changes_enabled()
)
result = check.execute()
assert len(result) == 0
def test_no_log_metric_filters_no_alerts_one_project(self):
logging_client = MagicMock()
monitoring_client = MagicMock()
with patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_gcp_provider(),
), patch(
"prowler.providers.gcp.services.logging.logging_log_metric_filter_and_alert_for_audit_configuration_changes_enabled.logging_log_metric_filter_and_alert_for_audit_configuration_changes_enabled.logging_client",
new=logging_client,
), patch(
"prowler.providers.gcp.services.logging.logging_log_metric_filter_and_alert_for_audit_configuration_changes_enabled.logging_log_metric_filter_and_alert_for_audit_configuration_changes_enabled.monitoring_client",
new=monitoring_client,
):
from prowler.providers.gcp.services.logging.logging_log_metric_filter_and_alert_for_audit_configuration_changes_enabled.logging_log_metric_filter_and_alert_for_audit_configuration_changes_enabled import (
logging_log_metric_filter_and_alert_for_audit_configuration_changes_enabled,
)
logging_client.metrics = []
logging_client.project_ids = [GCP_PROJECT_ID]
logging_client.region = GCP_EU1_LOCATION
monitoring_client.alert_policies = []
check = (
logging_log_metric_filter_and_alert_for_audit_configuration_changes_enabled()
)
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== f"There are no log metric filters or alerts associated in project {GCP_PROJECT_ID}."
)
assert result[0].resource_id == GCP_PROJECT_ID
assert result[0].resource_name == ""
assert result[0].project_id == GCP_PROJECT_ID
assert result[0].location == GCP_EU1_LOCATION
def test_log_metric_filters_no_alerts(self):
logging_client = MagicMock()
monitoring_client = MagicMock()
with patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_gcp_provider(),
), patch(
"prowler.providers.gcp.services.logging.logging_log_metric_filter_and_alert_for_audit_configuration_changes_enabled.logging_log_metric_filter_and_alert_for_audit_configuration_changes_enabled.logging_client",
new=logging_client,
), patch(
"prowler.providers.gcp.services.logging.logging_log_metric_filter_and_alert_for_audit_configuration_changes_enabled.logging_log_metric_filter_and_alert_for_audit_configuration_changes_enabled.monitoring_client",
new=monitoring_client,
):
from prowler.providers.gcp.services.logging.logging_log_metric_filter_and_alert_for_audit_configuration_changes_enabled.logging_log_metric_filter_and_alert_for_audit_configuration_changes_enabled import (
logging_log_metric_filter_and_alert_for_audit_configuration_changes_enabled,
)
from prowler.providers.gcp.services.logging.logging_service import Metric
logging_client.metrics = [
Metric(
name="metric_name",
type="custom.googleapis.com/invoice/paid/amount",
filter='protoPayload.methodName="SetIamPolicy" AND protoPayload.serviceData.policyDelta.auditConfigDeltas:*',
project_id=GCP_PROJECT_ID,
)
]
logging_client.project_ids = [GCP_PROJECT_ID]
logging_client.region = GCP_EU1_LOCATION
monitoring_client.alert_policies = []
check = (
logging_log_metric_filter_and_alert_for_audit_configuration_changes_enabled()
)
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== f"Log metric filter metric_name found but no alerts associated in project {GCP_PROJECT_ID}."
)
assert result[0].resource_id == "metric_name"
assert result[0].resource_name == "metric_name"
assert result[0].project_id == GCP_PROJECT_ID
assert result[0].location == GCP_EU1_LOCATION
def test_log_metric_filters_with_alerts(self):
logging_client = MagicMock()
monitoring_client = MagicMock()
with patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_gcp_provider(),
), patch(
"prowler.providers.gcp.services.logging.logging_log_metric_filter_and_alert_for_audit_configuration_changes_enabled.logging_log_metric_filter_and_alert_for_audit_configuration_changes_enabled.logging_client",
new=logging_client,
), patch(
"prowler.providers.gcp.services.logging.logging_log_metric_filter_and_alert_for_audit_configuration_changes_enabled.logging_log_metric_filter_and_alert_for_audit_configuration_changes_enabled.monitoring_client",
new=monitoring_client,
):
from prowler.providers.gcp.services.logging.logging_log_metric_filter_and_alert_for_audit_configuration_changes_enabled.logging_log_metric_filter_and_alert_for_audit_configuration_changes_enabled import (
logging_log_metric_filter_and_alert_for_audit_configuration_changes_enabled,
)
from prowler.providers.gcp.services.logging.logging_service import Metric
from prowler.providers.gcp.services.monitoring.monitoring_service import (
AlertPolicy,
)
logging_client.metrics = [
Metric(
name="metric_name",
type="metric_type",
filter='protoPayload.methodName="SetIamPolicy" AND protoPayload.serviceData.policyDelta.auditConfigDeltas:*',
project_id=GCP_PROJECT_ID,
)
]
logging_client.project_ids = [GCP_PROJECT_ID]
logging_client.region = GCP_EU1_LOCATION
monitoring_client.alert_policies = [
AlertPolicy(
name=f"projects/{GCP_PROJECT_ID}/alertPolicies/alert_policy",
display_name="alert_policy",
enabled=True,
filters=[
'metric.type = "logging.googleapis.com/user/metric_name"',
],
project_id=GCP_PROJECT_ID,
)
]
check = (
logging_log_metric_filter_and_alert_for_audit_configuration_changes_enabled()
)
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert (
result[0].status_extended
== f"Log metric filter metric_name found with alert policy alert_policy associated in project {GCP_PROJECT_ID}."
)
assert result[0].resource_id == "metric_name"
assert result[0].resource_name == "metric_name"
assert result[0].project_id == GCP_PROJECT_ID
assert result[0].location == GCP_EU1_LOCATION
tests/providers/gcp/services/logging/logging_log_metric_filter_and_alert_for_bucket_permission_changes_enabled/logging_log_metric_filter_and_alert_for_bucket_permission_changes_enabled_test.py
+183
View File
@@ -0,0 +1,183 @@
from unittest.mock import MagicMock, patch
from tests.providers.gcp.gcp_fixtures import (
GCP_EU1_LOCATION,
GCP_PROJECT_ID,
set_mocked_gcp_provider,
)
class Test_logging_log_metric_filter_and_alert_for_bucket_permission_changes_enabled:
def test_no_projects(self):
logging_client = MagicMock()
monitoring_client = MagicMock()
with patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_gcp_provider(),
), patch(
"prowler.providers.gcp.services.logging.logging_log_metric_filter_and_alert_for_bucket_permission_changes_enabled.logging_log_metric_filter_and_alert_for_bucket_permission_changes_enabled.logging_client",
new=logging_client,
), patch(
"prowler.providers.gcp.services.logging.logging_log_metric_filter_and_alert_for_bucket_permission_changes_enabled.logging_log_metric_filter_and_alert_for_bucket_permission_changes_enabled.monitoring_client",
new=monitoring_client,
):
from prowler.providers.gcp.services.logging.logging_log_metric_filter_and_alert_for_bucket_permission_changes_enabled.logging_log_metric_filter_and_alert_for_bucket_permission_changes_enabled import (
logging_log_metric_filter_and_alert_for_bucket_permission_changes_enabled,
)
logging_client.metrics = []
logging_client.project_ids = []
monitoring_client.alert_policies = []
check = (
logging_log_metric_filter_and_alert_for_bucket_permission_changes_enabled()
)
result = check.execute()
assert len(result) == 0
def test_no_log_metric_filters_no_alerts_one_project(self):
logging_client = MagicMock()
monitoring_client = MagicMock()
with patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_gcp_provider(),
), patch(
"prowler.providers.gcp.services.logging.logging_log_metric_filter_and_alert_for_bucket_permission_changes_enabled.logging_log_metric_filter_and_alert_for_bucket_permission_changes_enabled.logging_client",
new=logging_client,
), patch(
"prowler.providers.gcp.services.logging.logging_log_metric_filter_and_alert_for_bucket_permission_changes_enabled.logging_log_metric_filter_and_alert_for_bucket_permission_changes_enabled.monitoring_client",
new=monitoring_client,
):
from prowler.providers.gcp.services.logging.logging_log_metric_filter_and_alert_for_bucket_permission_changes_enabled.logging_log_metric_filter_and_alert_for_bucket_permission_changes_enabled import (
logging_log_metric_filter_and_alert_for_bucket_permission_changes_enabled,
)
logging_client.metrics = []
logging_client.project_ids = [GCP_PROJECT_ID]
logging_client.region = GCP_EU1_LOCATION
monitoring_client.alert_policies = []
check = (
logging_log_metric_filter_and_alert_for_bucket_permission_changes_enabled()
)
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== f"There are no log metric filters or alerts associated in project {GCP_PROJECT_ID}."
)
assert result[0].resource_id == GCP_PROJECT_ID
assert result[0].resource_name == ""
assert result[0].project_id == GCP_PROJECT_ID
assert result[0].location == GCP_EU1_LOCATION
def test_log_metric_filters_no_alerts(self):
logging_client = MagicMock()
monitoring_client = MagicMock()
with patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_gcp_provider(),
), patch(
"prowler.providers.gcp.services.logging.logging_log_metric_filter_and_alert_for_bucket_permission_changes_enabled.logging_log_metric_filter_and_alert_for_bucket_permission_changes_enabled.logging_client",
new=logging_client,
), patch(
"prowler.providers.gcp.services.logging.logging_log_metric_filter_and_alert_for_bucket_permission_changes_enabled.logging_log_metric_filter_and_alert_for_bucket_permission_changes_enabled.monitoring_client",
new=monitoring_client,
):
from prowler.providers.gcp.services.logging.logging_log_metric_filter_and_alert_for_bucket_permission_changes_enabled.logging_log_metric_filter_and_alert_for_bucket_permission_changes_enabled import (
logging_log_metric_filter_and_alert_for_bucket_permission_changes_enabled,
)
from prowler.providers.gcp.services.logging.logging_service import Metric
logging_client.metrics = [
Metric(
name="metric_name",
type="custom.googleapis.com/invoice/paid/amount",
filter='resource.type="gcs_bucket" AND protoPayload.methodName="storage.setIamPermissions"',
project_id=GCP_PROJECT_ID,
)
]
logging_client.project_ids = [GCP_PROJECT_ID]
logging_client.region = GCP_EU1_LOCATION
monitoring_client.alert_policies = []
check = (
logging_log_metric_filter_and_alert_for_bucket_permission_changes_enabled()
)
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== f"Log metric filter metric_name found but no alerts associated in project {GCP_PROJECT_ID}."
)
assert result[0].resource_id == "metric_name"
assert result[0].resource_name == "metric_name"
assert result[0].project_id == GCP_PROJECT_ID
assert result[0].location == GCP_EU1_LOCATION
def test_log_metric_filters_with_alerts(self):
logging_client = MagicMock()
monitoring_client = MagicMock()
with patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_gcp_provider(),
), patch(
"prowler.providers.gcp.services.logging.logging_log_metric_filter_and_alert_for_bucket_permission_changes_enabled.logging_log_metric_filter_and_alert_for_bucket_permission_changes_enabled.logging_client",
new=logging_client,
), patch(
"prowler.providers.gcp.services.logging.logging_log_metric_filter_and_alert_for_bucket_permission_changes_enabled.logging_log_metric_filter_and_alert_for_bucket_permission_changes_enabled.monitoring_client",
new=monitoring_client,
):
from prowler.providers.gcp.services.logging.logging_log_metric_filter_and_alert_for_bucket_permission_changes_enabled.logging_log_metric_filter_and_alert_for_bucket_permission_changes_enabled import (
logging_log_metric_filter_and_alert_for_bucket_permission_changes_enabled,
)
from prowler.providers.gcp.services.logging.logging_service import Metric
from prowler.providers.gcp.services.monitoring.monitoring_service import (
AlertPolicy,
)
logging_client.metrics = [
Metric(
name="metric_name",
type="metric_type",
filter='resource.type="gcs_bucket" AND protoPayload.methodName="storage.setIamPermissions"',
project_id=GCP_PROJECT_ID,
)
]
logging_client.project_ids = [GCP_PROJECT_ID]
logging_client.region = GCP_EU1_LOCATION
monitoring_client.alert_policies = [
AlertPolicy(
name=f"projects/{GCP_PROJECT_ID}/alertPolicies/alert_policy",
display_name="alert_policy",
enabled=True,
filters=[
'metric.type = "logging.googleapis.com/user/metric_name"',
],
project_id=GCP_PROJECT_ID,
)
]
check = (
logging_log_metric_filter_and_alert_for_bucket_permission_changes_enabled()
)
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert (
result[0].status_extended
== f"Log metric filter metric_name found with alert policy alert_policy associated in project {GCP_PROJECT_ID}."
)
assert result[0].resource_id == "metric_name"
assert result[0].resource_name == "metric_name"
assert result[0].project_id == GCP_PROJECT_ID
assert result[0].location == GCP_EU1_LOCATION
tests/providers/gcp/services/logging/logging_log_metric_filter_and_alert_for_custom_role_changes_enabled/logging_log_metric_filter_and_alert_for_custom_role_changes_enabled_test.py
+183
View File
@@ -0,0 +1,183 @@
from unittest.mock import MagicMock, patch
from tests.providers.gcp.gcp_fixtures import (
GCP_EU1_LOCATION,
GCP_PROJECT_ID,
set_mocked_gcp_provider,
)
class Test_logging_log_metric_filter_and_alert_for_custom_role_changes_enabled:
def test_no_projects(self):
logging_client = MagicMock()
monitoring_client = MagicMock()
with patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_gcp_provider(),
), patch(
"prowler.providers.gcp.services.logging.logging_log_metric_filter_and_alert_for_custom_role_changes_enabled.logging_log_metric_filter_and_alert_for_custom_role_changes_enabled.logging_client",
new=logging_client,
), patch(
"prowler.providers.gcp.services.logging.logging_log_metric_filter_and_alert_for_custom_role_changes_enabled.logging_log_metric_filter_and_alert_for_custom_role_changes_enabled.monitoring_client",
new=monitoring_client,
):
from prowler.providers.gcp.services.logging.logging_log_metric_filter_and_alert_for_custom_role_changes_enabled.logging_log_metric_filter_and_alert_for_custom_role_changes_enabled import (
logging_log_metric_filter_and_alert_for_custom_role_changes_enabled,
)
logging_client.metrics = []
logging_client.project_ids = []
monitoring_client.alert_policies = []
check = (
logging_log_metric_filter_and_alert_for_custom_role_changes_enabled()
)
result = check.execute()
assert len(result) == 0
def test_no_log_metric_filters_no_alerts_one_project(self):
logging_client = MagicMock()
monitoring_client = MagicMock()
with patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_gcp_provider(),
), patch(
"prowler.providers.gcp.services.logging.logging_log_metric_filter_and_alert_for_custom_role_changes_enabled.logging_log_metric_filter_and_alert_for_custom_role_changes_enabled.logging_client",
new=logging_client,
), patch(
"prowler.providers.gcp.services.logging.logging_log_metric_filter_and_alert_for_custom_role_changes_enabled.logging_log_metric_filter_and_alert_for_custom_role_changes_enabled.monitoring_client",
new=monitoring_client,
):
from prowler.providers.gcp.services.logging.logging_log_metric_filter_and_alert_for_custom_role_changes_enabled.logging_log_metric_filter_and_alert_for_custom_role_changes_enabled import (
logging_log_metric_filter_and_alert_for_custom_role_changes_enabled,
)
logging_client.metrics = []
logging_client.project_ids = [GCP_PROJECT_ID]
logging_client.region = GCP_EU1_LOCATION
monitoring_client.alert_policies = []
check = (
logging_log_metric_filter_and_alert_for_custom_role_changes_enabled()
)
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== f"There are no log metric filters or alerts associated in project {GCP_PROJECT_ID}."
)
assert result[0].resource_id == GCP_PROJECT_ID
assert result[0].resource_name == ""
assert result[0].project_id == GCP_PROJECT_ID
assert result[0].location == GCP_EU1_LOCATION
def test_log_metric_filters_no_alerts(self):
logging_client = MagicMock()
monitoring_client = MagicMock()
with patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_gcp_provider(),
), patch(
"prowler.providers.gcp.services.logging.logging_log_metric_filter_and_alert_for_custom_role_changes_enabled.logging_log_metric_filter_and_alert_for_custom_role_changes_enabled.logging_client",
new=logging_client,
), patch(
"prowler.providers.gcp.services.logging.logging_log_metric_filter_and_alert_for_custom_role_changes_enabled.logging_log_metric_filter_and_alert_for_custom_role_changes_enabled.monitoring_client",
new=monitoring_client,
):
from prowler.providers.gcp.services.logging.logging_log_metric_filter_and_alert_for_custom_role_changes_enabled.logging_log_metric_filter_and_alert_for_custom_role_changes_enabled import (
logging_log_metric_filter_and_alert_for_custom_role_changes_enabled,
)
from prowler.providers.gcp.services.logging.logging_service import Metric
logging_client.metrics = [
Metric(
name="metric_name",
type="custom.googleapis.com/invoice/paid/amount",
filter='resource.type="iam_role" AND (protoPayload.methodName="google.iam.admin.v1.CreateRole" OR protoPayload.methodName="google.iam.admin.v1.DeleteRole" OR protoPayload.methodName="google.iam.admin.v1.UpdateRole")',
project_id=GCP_PROJECT_ID,
)
]
logging_client.project_ids = [GCP_PROJECT_ID]
logging_client.region = GCP_EU1_LOCATION
monitoring_client.alert_policies = []
check = (
logging_log_metric_filter_and_alert_for_custom_role_changes_enabled()
)
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== f"Log metric filter metric_name found but no alerts associated in project {GCP_PROJECT_ID}."
)
assert result[0].resource_id == "metric_name"
assert result[0].resource_name == "metric_name"
assert result[0].project_id == GCP_PROJECT_ID
assert result[0].location == GCP_EU1_LOCATION
def test_log_metric_filters_with_alerts(self):
logging_client = MagicMock()
monitoring_client = MagicMock()
with patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_gcp_provider(),
), patch(
"prowler.providers.gcp.services.logging.logging_log_metric_filter_and_alert_for_custom_role_changes_enabled.logging_log_metric_filter_and_alert_for_custom_role_changes_enabled.logging_client",
new=logging_client,
), patch(
"prowler.providers.gcp.services.logging.logging_log_metric_filter_and_alert_for_custom_role_changes_enabled.logging_log_metric_filter_and_alert_for_custom_role_changes_enabled.monitoring_client",
new=monitoring_client,
):
from prowler.providers.gcp.services.logging.logging_log_metric_filter_and_alert_for_custom_role_changes_enabled.logging_log_metric_filter_and_alert_for_custom_role_changes_enabled import (
logging_log_metric_filter_and_alert_for_custom_role_changes_enabled,
)
from prowler.providers.gcp.services.logging.logging_service import Metric
from prowler.providers.gcp.services.monitoring.monitoring_service import (
AlertPolicy,
)
logging_client.metrics = [
Metric(
name="metric_name",
type="metric_type",
filter='resource.type="iam_role" AND (protoPayload.methodName="google.iam.admin.v1.CreateRole" OR protoPayload.methodName="google.iam.admin.v1.DeleteRole" OR protoPayload.methodName="google.iam.admin.v1.UpdateRole")',
project_id=GCP_PROJECT_ID,
)
]
logging_client.project_ids = [GCP_PROJECT_ID]
logging_client.region = GCP_EU1_LOCATION
monitoring_client.alert_policies = [
AlertPolicy(
name=f"projects/{GCP_PROJECT_ID}/alertPolicies/alert_policy",
display_name="alert_policy",
enabled=True,
filters=[
'metric.type = "logging.googleapis.com/user/metric_name"',
],
project_id=GCP_PROJECT_ID,
)
]
check = (
logging_log_metric_filter_and_alert_for_custom_role_changes_enabled()
)
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert (
result[0].status_extended
== f"Log metric filter metric_name found with alert policy alert_policy associated in project {GCP_PROJECT_ID}."
)
assert result[0].resource_id == "metric_name"
assert result[0].resource_name == "metric_name"
assert result[0].project_id == GCP_PROJECT_ID
assert result[0].location == GCP_EU1_LOCATION
tests/providers/gcp/services/logging/logging_log_metric_filter_and_alert_for_project_ownership_changes_enabled/logging_log_metric_filter_and_alert_for_project_ownership_changes_enabled_test.py
+183
View File
@@ -0,0 +1,183 @@
from unittest.mock import MagicMock, patch
from tests.providers.gcp.gcp_fixtures import (
GCP_EU1_LOCATION,
GCP_PROJECT_ID,
set_mocked_gcp_provider,
)
class Test_logging_log_metric_filter_and_alert_for_project_ownership_changes_enabled:
def test_no_projects(self):
logging_client = MagicMock()
monitoring_client = MagicMock()
with patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_gcp_provider(),
), patch(
"prowler.providers.gcp.services.logging.logging_log_metric_filter_and_alert_for_project_ownership_changes_enabled.logging_log_metric_filter_and_alert_for_project_ownership_changes_enabled.logging_client",
new=logging_client,
), patch(
"prowler.providers.gcp.services.logging.logging_log_metric_filter_and_alert_for_project_ownership_changes_enabled.logging_log_metric_filter_and_alert_for_project_ownership_changes_enabled.monitoring_client",
new=monitoring_client,
):
from prowler.providers.gcp.services.logging.logging_log_metric_filter_and_alert_for_project_ownership_changes_enabled.logging_log_metric_filter_and_alert_for_project_ownership_changes_enabled import (
logging_log_metric_filter_and_alert_for_project_ownership_changes_enabled,
)
logging_client.metrics = []
logging_client.project_ids = []
monitoring_client.alert_policies = []
check = (
logging_log_metric_filter_and_alert_for_project_ownership_changes_enabled()
)
result = check.execute()
assert len(result) == 0
def test_no_log_metric_filters_no_alerts_one_project(self):
logging_client = MagicMock()
monitoring_client = MagicMock()
with patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_gcp_provider(),
), patch(
"prowler.providers.gcp.services.logging.logging_log_metric_filter_and_alert_for_project_ownership_changes_enabled.logging_log_metric_filter_and_alert_for_project_ownership_changes_enabled.logging_client",
new=logging_client,
), patch(
"prowler.providers.gcp.services.logging.logging_log_metric_filter_and_alert_for_project_ownership_changes_enabled.logging_log_metric_filter_and_alert_for_project_ownership_changes_enabled.monitoring_client",
new=monitoring_client,
):
from prowler.providers.gcp.services.logging.logging_log_metric_filter_and_alert_for_project_ownership_changes_enabled.logging_log_metric_filter_and_alert_for_project_ownership_changes_enabled import (
logging_log_metric_filter_and_alert_for_project_ownership_changes_enabled,
)
logging_client.metrics = []
logging_client.project_ids = [GCP_PROJECT_ID]
logging_client.region = GCP_EU1_LOCATION
monitoring_client.alert_policies = []
check = (
logging_log_metric_filter_and_alert_for_project_ownership_changes_enabled()
)
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== f"There are no log metric filters or alerts associated in project {GCP_PROJECT_ID}."
)
assert result[0].resource_id == GCP_PROJECT_ID
assert result[0].resource_name == ""
assert result[0].project_id == GCP_PROJECT_ID
assert result[0].location == GCP_EU1_LOCATION
def test_log_metric_filters_no_alerts(self):
logging_client = MagicMock()
monitoring_client = MagicMock()
with patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_gcp_provider(),
), patch(
"prowler.providers.gcp.services.logging.logging_log_metric_filter_and_alert_for_project_ownership_changes_enabled.logging_log_metric_filter_and_alert_for_project_ownership_changes_enabled.logging_client",
new=logging_client,
), patch(
"prowler.providers.gcp.services.logging.logging_log_metric_filter_and_alert_for_project_ownership_changes_enabled.logging_log_metric_filter_and_alert_for_project_ownership_changes_enabled.monitoring_client",
new=monitoring_client,
):
from prowler.providers.gcp.services.logging.logging_log_metric_filter_and_alert_for_project_ownership_changes_enabled.logging_log_metric_filter_and_alert_for_project_ownership_changes_enabled import (
logging_log_metric_filter_and_alert_for_project_ownership_changes_enabled,
)
from prowler.providers.gcp.services.logging.logging_service import Metric
logging_client.metrics = [
Metric(
name="metric_name",
type="custom.googleapis.com/invoice/paid/amount",
filter='(protoPayload.serviceName="cloudresourcemanager.googleapis.com") AND (ProjectOwnership OR projectOwnerInvitee) OR (protoPayload.serviceData.policyDelta.bindingDeltas.action="REMOVE" AND protoPayload.serviceData.policyDelta.bindingDeltas.role="roles/owner") OR (protoPayload.serviceData.policyDelta.bindingDeltas.action="ADD" AND protoPayload.serviceData.policyDelta.bindingDeltas.role="roles/owner")',
project_id=GCP_PROJECT_ID,
)
]
logging_client.project_ids = [GCP_PROJECT_ID]
logging_client.region = GCP_EU1_LOCATION
monitoring_client.alert_policies = []
check = (
logging_log_metric_filter_and_alert_for_project_ownership_changes_enabled()
)
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== f"Log metric filter metric_name found but no alerts associated in project {GCP_PROJECT_ID}."
)
assert result[0].resource_id == "metric_name"
assert result[0].resource_name == "metric_name"
assert result[0].project_id == GCP_PROJECT_ID
assert result[0].location == GCP_EU1_LOCATION
def test_log_metric_filters_with_alerts(self):
logging_client = MagicMock()
monitoring_client = MagicMock()
with patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_gcp_provider(),
), patch(
"prowler.providers.gcp.services.logging.logging_log_metric_filter_and_alert_for_project_ownership_changes_enabled.logging_log_metric_filter_and_alert_for_project_ownership_changes_enabled.logging_client",
new=logging_client,
), patch(
"prowler.providers.gcp.services.logging.logging_log_metric_filter_and_alert_for_project_ownership_changes_enabled.logging_log_metric_filter_and_alert_for_project_ownership_changes_enabled.monitoring_client",
new=monitoring_client,
):
from prowler.providers.gcp.services.logging.logging_log_metric_filter_and_alert_for_project_ownership_changes_enabled.logging_log_metric_filter_and_alert_for_project_ownership_changes_enabled import (
logging_log_metric_filter_and_alert_for_project_ownership_changes_enabled,
)
from prowler.providers.gcp.services.logging.logging_service import Metric
from prowler.providers.gcp.services.monitoring.monitoring_service import (
AlertPolicy,
)
logging_client.metrics = [
Metric(
name="metric_name",
type="metric_type",
filter='(protoPayload.serviceName="cloudresourcemanager.googleapis.com") AND (ProjectOwnership OR projectOwnerInvitee) OR (protoPayload.serviceData.policyDelta.bindingDeltas.action="REMOVE" AND protoPayload.serviceData.policyDelta.bindingDeltas.role="roles/owner") OR (protoPayload.serviceData.policyDelta.bindingDeltas.action="ADD" AND protoPayload.serviceData.policyDelta.bindingDeltas.role="roles/owner")',
project_id=GCP_PROJECT_ID,
)
]
logging_client.project_ids = [GCP_PROJECT_ID]
logging_client.region = GCP_EU1_LOCATION
monitoring_client.alert_policies = [
AlertPolicy(
name=f"projects/{GCP_PROJECT_ID}/alertPolicies/alert_policy",
display_name="alert_policy",
enabled=True,
filters=[
'metric.type = "logging.googleapis.com/user/metric_name"',
],
project_id=GCP_PROJECT_ID,
)
]
check = (
logging_log_metric_filter_and_alert_for_project_ownership_changes_enabled()
)
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert (
result[0].status_extended
== f"Log metric filter metric_name found with alert policy alert_policy associated in project {GCP_PROJECT_ID}."
)
assert result[0].resource_id == "metric_name"
assert result[0].resource_name == "metric_name"
assert result[0].project_id == GCP_PROJECT_ID
assert result[0].location == GCP_EU1_LOCATION
tests/providers/gcp/services/logging/logging_log_metric_filter_and_alert_for_sql_instance_configuration_changes_enabled/logging_log_metric_filter_and_alert_for_sql_instance_configuration_changes_enabled_test.py
+183
View File
@@ -0,0 +1,183 @@
from unittest.mock import MagicMock, patch
from tests.providers.gcp.gcp_fixtures import (
GCP_EU1_LOCATION,
GCP_PROJECT_ID,
set_mocked_gcp_provider,
)
class Test_logging_log_metric_filter_and_alert_for_sql_instance_configuration_changes_enabled:
def test_no_projects(self):
logging_client = MagicMock()
monitoring_client = MagicMock()
with patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_gcp_provider(),
), patch(
"prowler.providers.gcp.services.logging.logging_log_metric_filter_and_alert_for_sql_instance_configuration_changes_enabled.logging_log_metric_filter_and_alert_for_sql_instance_configuration_changes_enabled.logging_client",
new=logging_client,
), patch(
"prowler.providers.gcp.services.logging.logging_log_metric_filter_and_alert_for_sql_instance_configuration_changes_enabled.logging_log_metric_filter_and_alert_for_sql_instance_configuration_changes_enabled.monitoring_client",
new=monitoring_client,
):
from prowler.providers.gcp.services.logging.logging_log_metric_filter_and_alert_for_sql_instance_configuration_changes_enabled.logging_log_metric_filter_and_alert_for_sql_instance_configuration_changes_enabled import (
logging_log_metric_filter_and_alert_for_sql_instance_configuration_changes_enabled,
)
logging_client.metrics = []
logging_client.project_ids = []
monitoring_client.alert_policies = []
check = (
logging_log_metric_filter_and_alert_for_sql_instance_configuration_changes_enabled()
)
result = check.execute()
assert len(result) == 0
def test_no_log_metric_filters_no_alerts_one_project(self):
logging_client = MagicMock()
monitoring_client = MagicMock()
with patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_gcp_provider(),
), patch(
"prowler.providers.gcp.services.logging.logging_log_metric_filter_and_alert_for_sql_instance_configuration_changes_enabled.logging_log_metric_filter_and_alert_for_sql_instance_configuration_changes_enabled.logging_client",
new=logging_client,
), patch(
"prowler.providers.gcp.services.logging.logging_log_metric_filter_and_alert_for_sql_instance_configuration_changes_enabled.logging_log_metric_filter_and_alert_for_sql_instance_configuration_changes_enabled.monitoring_client",
new=monitoring_client,
):
from prowler.providers.gcp.services.logging.logging_log_metric_filter_and_alert_for_sql_instance_configuration_changes_enabled.logging_log_metric_filter_and_alert_for_sql_instance_configuration_changes_enabled import (
logging_log_metric_filter_and_alert_for_sql_instance_configuration_changes_enabled,
)
logging_client.metrics = []
logging_client.project_ids = [GCP_PROJECT_ID]
logging_client.region = GCP_EU1_LOCATION
monitoring_client.alert_policies = []
check = (
logging_log_metric_filter_and_alert_for_sql_instance_configuration_changes_enabled()
)
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== f"There are no log metric filters or alerts associated in project {GCP_PROJECT_ID}."
)
assert result[0].resource_id == GCP_PROJECT_ID
assert result[0].resource_name == ""
assert result[0].project_id == GCP_PROJECT_ID
assert result[0].location == GCP_EU1_LOCATION
def test_log_metric_filters_no_alerts(self):
logging_client = MagicMock()
monitoring_client = MagicMock()
with patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_gcp_provider(),
), patch(
"prowler.providers.gcp.services.logging.logging_log_metric_filter_and_alert_for_sql_instance_configuration_changes_enabled.logging_log_metric_filter_and_alert_for_sql_instance_configuration_changes_enabled.logging_client",
new=logging_client,
), patch(
"prowler.providers.gcp.services.logging.logging_log_metric_filter_and_alert_for_sql_instance_configuration_changes_enabled.logging_log_metric_filter_and_alert_for_sql_instance_configuration_changes_enabled.monitoring_client",
new=monitoring_client,
):
from prowler.providers.gcp.services.logging.logging_log_metric_filter_and_alert_for_sql_instance_configuration_changes_enabled.logging_log_metric_filter_and_alert_for_sql_instance_configuration_changes_enabled import (
logging_log_metric_filter_and_alert_for_sql_instance_configuration_changes_enabled,
)
from prowler.providers.gcp.services.logging.logging_service import Metric
logging_client.metrics = [
Metric(
name="metric_name",
type="custom.googleapis.com/invoice/paid/amount",
filter='protoPayload.methodName="cloudsql.instances.update"',
project_id=GCP_PROJECT_ID,
)
]
logging_client.project_ids = [GCP_PROJECT_ID]
logging_client.region = GCP_EU1_LOCATION
monitoring_client.alert_policies = []
check = (
logging_log_metric_filter_and_alert_for_sql_instance_configuration_changes_enabled()
)
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== f"Log metric filter metric_name found but no alerts associated in project {GCP_PROJECT_ID}."
)
assert result[0].resource_id == "metric_name"
assert result[0].resource_name == "metric_name"
assert result[0].project_id == GCP_PROJECT_ID
assert result[0].location == GCP_EU1_LOCATION
def test_log_metric_filters_with_alerts(self):
logging_client = MagicMock()
monitoring_client = MagicMock()
with patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_gcp_provider(),
), patch(
"prowler.providers.gcp.services.logging.logging_log_metric_filter_and_alert_for_sql_instance_configuration_changes_enabled.logging_log_metric_filter_and_alert_for_sql_instance_configuration_changes_enabled.logging_client",
new=logging_client,
), patch(
"prowler.providers.gcp.services.logging.logging_log_metric_filter_and_alert_for_sql_instance_configuration_changes_enabled.logging_log_metric_filter_and_alert_for_sql_instance_configuration_changes_enabled.monitoring_client",
new=monitoring_client,
):
from prowler.providers.gcp.services.logging.logging_log_metric_filter_and_alert_for_sql_instance_configuration_changes_enabled.logging_log_metric_filter_and_alert_for_sql_instance_configuration_changes_enabled import (
logging_log_metric_filter_and_alert_for_sql_instance_configuration_changes_enabled,
)
from prowler.providers.gcp.services.logging.logging_service import Metric
from prowler.providers.gcp.services.monitoring.monitoring_service import (
AlertPolicy,
)
logging_client.metrics = [
Metric(
name="metric_name",
type="metric_type",
filter='protoPayload.methodName="cloudsql.instances.update"',
project_id=GCP_PROJECT_ID,
)
]
logging_client.project_ids = [GCP_PROJECT_ID]
logging_client.region = GCP_EU1_LOCATION
monitoring_client.alert_policies = [
AlertPolicy(
name=f"projects/{GCP_PROJECT_ID}/alertPolicies/alert_policy",
display_name="alert_policy",
enabled=True,
filters=[
'metric.type = "logging.googleapis.com/user/metric_name"',
],
project_id=GCP_PROJECT_ID,
)
]
check = (
logging_log_metric_filter_and_alert_for_sql_instance_configuration_changes_enabled()
)
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert (
result[0].status_extended
== f"Log metric filter metric_name found with alert policy alert_policy associated in project {GCP_PROJECT_ID}."
)
assert result[0].resource_id == "metric_name"
assert result[0].resource_name == "metric_name"
assert result[0].project_id == GCP_PROJECT_ID
assert result[0].location == GCP_EU1_LOCATION
tests/providers/gcp/services/logging/logging_log_metric_filter_and_alert_for_vpc_firewall_rule_changes_enabled/logging_log_metric_filter_and_alert_for_vpc_firewall_rule_changes_enabled_test.py
+183
View File
@@ -0,0 +1,183 @@
from unittest.mock import MagicMock, patch
from tests.providers.gcp.gcp_fixtures import (
GCP_EU1_LOCATION,
GCP_PROJECT_ID,
set_mocked_gcp_provider,
)
class Test_logging_log_metric_filter_and_alert_for_vpc_firewall_rule_changes_enabled:
def test_no_projects(self):
logging_client = MagicMock()
monitoring_client = MagicMock()
with patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_gcp_provider(),
), patch(
"prowler.providers.gcp.services.logging.logging_log_metric_filter_and_alert_for_vpc_firewall_rule_changes_enabled.logging_log_metric_filter_and_alert_for_vpc_firewall_rule_changes_enabled.logging_client",
new=logging_client,
), patch(
"prowler.providers.gcp.services.logging.logging_log_metric_filter_and_alert_for_vpc_firewall_rule_changes_enabled.logging_log_metric_filter_and_alert_for_vpc_firewall_rule_changes_enabled.monitoring_client",
new=monitoring_client,
):
from prowler.providers.gcp.services.logging.logging_log_metric_filter_and_alert_for_vpc_firewall_rule_changes_enabled.logging_log_metric_filter_and_alert_for_vpc_firewall_rule_changes_enabled import (
logging_log_metric_filter_and_alert_for_vpc_firewall_rule_changes_enabled,
)
logging_client.metrics = []
logging_client.project_ids = []
monitoring_client.alert_policies = []
check = (
logging_log_metric_filter_and_alert_for_vpc_firewall_rule_changes_enabled()
)
result = check.execute()
assert len(result) == 0
def test_no_log_metric_filters_no_alerts_one_project(self):
logging_client = MagicMock()
monitoring_client = MagicMock()
with patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_gcp_provider(),
), patch(
"prowler.providers.gcp.services.logging.logging_log_metric_filter_and_alert_for_vpc_firewall_rule_changes_enabled.logging_log_metric_filter_and_alert_for_vpc_firewall_rule_changes_enabled.logging_client",
new=logging_client,
), patch(
"prowler.providers.gcp.services.logging.logging_log_metric_filter_and_alert_for_vpc_firewall_rule_changes_enabled.logging_log_metric_filter_and_alert_for_vpc_firewall_rule_changes_enabled.monitoring_client",
new=monitoring_client,
):
from prowler.providers.gcp.services.logging.logging_log_metric_filter_and_alert_for_vpc_firewall_rule_changes_enabled.logging_log_metric_filter_and_alert_for_vpc_firewall_rule_changes_enabled import (
logging_log_metric_filter_and_alert_for_vpc_firewall_rule_changes_enabled,
)
logging_client.metrics = []
logging_client.project_ids = [GCP_PROJECT_ID]
logging_client.region = GCP_EU1_LOCATION
monitoring_client.alert_policies = []
check = (
logging_log_metric_filter_and_alert_for_vpc_firewall_rule_changes_enabled()
)
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== f"There are no log metric filters or alerts associated in project {GCP_PROJECT_ID}."
)
assert result[0].resource_id == GCP_PROJECT_ID
assert result[0].resource_name == ""
assert result[0].project_id == GCP_PROJECT_ID
assert result[0].location == GCP_EU1_LOCATION
def test_log_metric_filters_no_alerts(self):
logging_client = MagicMock()
monitoring_client = MagicMock()
with patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_gcp_provider(),
), patch(
"prowler.providers.gcp.services.logging.logging_log_metric_filter_and_alert_for_vpc_firewall_rule_changes_enabled.logging_log_metric_filter_and_alert_for_vpc_firewall_rule_changes_enabled.logging_client",
new=logging_client,
), patch(
"prowler.providers.gcp.services.logging.logging_log_metric_filter_and_alert_for_vpc_firewall_rule_changes_enabled.logging_log_metric_filter_and_alert_for_vpc_firewall_rule_changes_enabled.monitoring_client",
new=monitoring_client,
):
from prowler.providers.gcp.services.logging.logging_log_metric_filter_and_alert_for_vpc_firewall_rule_changes_enabled.logging_log_metric_filter_and_alert_for_vpc_firewall_rule_changes_enabled import (
logging_log_metric_filter_and_alert_for_vpc_firewall_rule_changes_enabled,
)
from prowler.providers.gcp.services.logging.logging_service import Metric
logging_client.metrics = [
Metric(
name="metric_name",
type="custom.googleapis.com/invoice/paid/amount",
filter='resource.type="gce_firewall_rule" AND (protoPayload.methodName:"compute.firewalls.patch" OR protoPayload.methodName:"compute.firewalls.insert" OR protoPayload.methodName:"compute.firewalls.delete")',
project_id=GCP_PROJECT_ID,
)
]
logging_client.project_ids = [GCP_PROJECT_ID]
logging_client.region = GCP_EU1_LOCATION
monitoring_client.alert_policies = []
check = (
logging_log_metric_filter_and_alert_for_vpc_firewall_rule_changes_enabled()
)
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== f"Log metric filter metric_name found but no alerts associated in project {GCP_PROJECT_ID}."
)
assert result[0].resource_id == "metric_name"
assert result[0].resource_name == "metric_name"
assert result[0].project_id == GCP_PROJECT_ID
assert result[0].location == GCP_EU1_LOCATION
def test_log_metric_filters_with_alerts(self):
logging_client = MagicMock()
monitoring_client = MagicMock()
with patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_gcp_provider(),
), patch(
"prowler.providers.gcp.services.logging.logging_log_metric_filter_and_alert_for_vpc_firewall_rule_changes_enabled.logging_log_metric_filter_and_alert_for_vpc_firewall_rule_changes_enabled.logging_client",
new=logging_client,
), patch(
"prowler.providers.gcp.services.logging.logging_log_metric_filter_and_alert_for_vpc_firewall_rule_changes_enabled.logging_log_metric_filter_and_alert_for_vpc_firewall_rule_changes_enabled.monitoring_client",
new=monitoring_client,
):
from prowler.providers.gcp.services.logging.logging_log_metric_filter_and_alert_for_vpc_firewall_rule_changes_enabled.logging_log_metric_filter_and_alert_for_vpc_firewall_rule_changes_enabled import (
logging_log_metric_filter_and_alert_for_vpc_firewall_rule_changes_enabled,
)
from prowler.providers.gcp.services.logging.logging_service import Metric
from prowler.providers.gcp.services.monitoring.monitoring_service import (
AlertPolicy,
)
logging_client.metrics = [
Metric(
name="metric_name",
type="metric_type",
filter='resource.type="gce_firewall_rule" AND (protoPayload.methodName:"compute.firewalls.patch" OR protoPayload.methodName:"compute.firewalls.insert" OR protoPayload.methodName:"compute.firewalls.delete")',
project_id=GCP_PROJECT_ID,
)
]
logging_client.project_ids = [GCP_PROJECT_ID]
logging_client.region = GCP_EU1_LOCATION
monitoring_client.alert_policies = [
AlertPolicy(
name=f"projects/{GCP_PROJECT_ID}/alertPolicies/alert_policy",
display_name="alert_policy",
enabled=True,
filters=[
'metric.type = "logging.googleapis.com/user/metric_name"',
],
project_id=GCP_PROJECT_ID,
)
]
check = (
logging_log_metric_filter_and_alert_for_vpc_firewall_rule_changes_enabled()
)
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert (
result[0].status_extended
== f"Log metric filter metric_name found with alert policy alert_policy associated in project {GCP_PROJECT_ID}."
)
assert result[0].resource_id == "metric_name"
assert result[0].resource_name == "metric_name"
assert result[0].project_id == GCP_PROJECT_ID
assert result[0].location == GCP_EU1_LOCATION
tests/providers/gcp/services/logging/logging_log_metric_filter_and_alert_for_vpc_network_changes_enabled/logging_log_metric_filter_and_alert_for_vpc_network_changes_enabled_test.py
+183
View File
@@ -0,0 +1,183 @@
from unittest.mock import MagicMock, patch
from tests.providers.gcp.gcp_fixtures import (
GCP_EU1_LOCATION,
GCP_PROJECT_ID,
set_mocked_gcp_provider,
)
class Test_logging_log_metric_filter_and_alert_for_vpc_network_changes_enabled:
def test_no_projects(self):
logging_client = MagicMock()
monitoring_client = MagicMock()
with patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_gcp_provider(),
), patch(
"prowler.providers.gcp.services.logging.logging_log_metric_filter_and_alert_for_vpc_network_changes_enabled.logging_log_metric_filter_and_alert_for_vpc_network_changes_enabled.logging_client",
new=logging_client,
), patch(
"prowler.providers.gcp.services.logging.logging_log_metric_filter_and_alert_for_vpc_network_changes_enabled.logging_log_metric_filter_and_alert_for_vpc_network_changes_enabled.monitoring_client",
new=monitoring_client,
):
from prowler.providers.gcp.services.logging.logging_log_metric_filter_and_alert_for_vpc_network_changes_enabled.logging_log_metric_filter_and_alert_for_vpc_network_changes_enabled import (
logging_log_metric_filter_and_alert_for_vpc_network_changes_enabled,
)
logging_client.metrics = []
logging_client.project_ids = []
monitoring_client.alert_policies = []
check = (
logging_log_metric_filter_and_alert_for_vpc_network_changes_enabled()
)
result = check.execute()
assert len(result) == 0
def test_no_log_metric_filters_no_alerts_one_project(self):
logging_client = MagicMock()
monitoring_client = MagicMock()
with patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_gcp_provider(),
), patch(
"prowler.providers.gcp.services.logging.logging_log_metric_filter_and_alert_for_vpc_network_changes_enabled.logging_log_metric_filter_and_alert_for_vpc_network_changes_enabled.logging_client",
new=logging_client,
), patch(
"prowler.providers.gcp.services.logging.logging_log_metric_filter_and_alert_for_vpc_network_changes_enabled.logging_log_metric_filter_and_alert_for_vpc_network_changes_enabled.monitoring_client",
new=monitoring_client,
):
from prowler.providers.gcp.services.logging.logging_log_metric_filter_and_alert_for_vpc_network_changes_enabled.logging_log_metric_filter_and_alert_for_vpc_network_changes_enabled import (
logging_log_metric_filter_and_alert_for_vpc_network_changes_enabled,
)
logging_client.metrics = []
logging_client.project_ids = [GCP_PROJECT_ID]
logging_client.region = GCP_EU1_LOCATION
monitoring_client.alert_policies = []
check = (
logging_log_metric_filter_and_alert_for_vpc_network_changes_enabled()
)
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== f"There are no log metric filters or alerts associated in project {GCP_PROJECT_ID}."
)
assert result[0].resource_id == GCP_PROJECT_ID
assert result[0].resource_name == ""
assert result[0].project_id == GCP_PROJECT_ID
assert result[0].location == GCP_EU1_LOCATION
def test_log_metric_filters_no_alerts(self):
logging_client = MagicMock()
monitoring_client = MagicMock()
with patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_gcp_provider(),
), patch(
"prowler.providers.gcp.services.logging.logging_log_metric_filter_and_alert_for_vpc_network_changes_enabled.logging_log_metric_filter_and_alert_for_vpc_network_changes_enabled.logging_client",
new=logging_client,
), patch(
"prowler.providers.gcp.services.logging.logging_log_metric_filter_and_alert_for_vpc_network_changes_enabled.logging_log_metric_filter_and_alert_for_vpc_network_changes_enabled.monitoring_client",
new=monitoring_client,
):
from prowler.providers.gcp.services.logging.logging_log_metric_filter_and_alert_for_vpc_network_changes_enabled.logging_log_metric_filter_and_alert_for_vpc_network_changes_enabled import (
logging_log_metric_filter_and_alert_for_vpc_network_changes_enabled,
)
from prowler.providers.gcp.services.logging.logging_service import Metric
logging_client.metrics = [
Metric(
name="metric_name",
type="custom.googleapis.com/invoice/paid/amount",
filter='resource.type="gce_network" AND (protoPayload.methodName:"compute.networks.insert" OR protoPayload.methodName:"compute.networks.patch" OR protoPayload.methodName:"compute.networks.delete" OR protoPayload.methodName:"compute.networks.removePeering" OR protoPayload.methodName:"compute.networks.addPeering")',
project_id=GCP_PROJECT_ID,
)
]
logging_client.project_ids = [GCP_PROJECT_ID]
logging_client.region = GCP_EU1_LOCATION
monitoring_client.alert_policies = []
check = (
logging_log_metric_filter_and_alert_for_vpc_network_changes_enabled()
)
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== f"Log metric filter metric_name found but no alerts associated in project {GCP_PROJECT_ID}."
)
assert result[0].resource_id == "metric_name"
assert result[0].resource_name == "metric_name"
assert result[0].project_id == GCP_PROJECT_ID
assert result[0].location == GCP_EU1_LOCATION
def test_log_metric_filters_with_alerts(self):
logging_client = MagicMock()
monitoring_client = MagicMock()
with patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_gcp_provider(),
), patch(
"prowler.providers.gcp.services.logging.logging_log_metric_filter_and_alert_for_vpc_network_changes_enabled.logging_log_metric_filter_and_alert_for_vpc_network_changes_enabled.logging_client",
new=logging_client,
), patch(
"prowler.providers.gcp.services.logging.logging_log_metric_filter_and_alert_for_vpc_network_changes_enabled.logging_log_metric_filter_and_alert_for_vpc_network_changes_enabled.monitoring_client",
new=monitoring_client,
):
from prowler.providers.gcp.services.logging.logging_log_metric_filter_and_alert_for_vpc_network_changes_enabled.logging_log_metric_filter_and_alert_for_vpc_network_changes_enabled import (
logging_log_metric_filter_and_alert_for_vpc_network_changes_enabled,
)
from prowler.providers.gcp.services.logging.logging_service import Metric
from prowler.providers.gcp.services.monitoring.monitoring_service import (
AlertPolicy,
)
logging_client.metrics = [
Metric(
name="metric_name",
type="metric_type",
filter='resource.type="gce_network" AND (protoPayload.methodName:"compute.networks.insert" OR protoPayload.methodName:"compute.networks.patch" OR protoPayload.methodName:"compute.networks.delete" OR protoPayload.methodName:"compute.networks.removePeering" OR protoPayload.methodName:"compute.networks.addPeering")',
project_id=GCP_PROJECT_ID,
)
]
logging_client.project_ids = [GCP_PROJECT_ID]
logging_client.region = GCP_EU1_LOCATION
monitoring_client.alert_policies = [
AlertPolicy(
name=f"projects/{GCP_PROJECT_ID}/alertPolicies/alert_policy",
display_name="alert_policy",
enabled=True,
filters=[
'metric.type = "logging.googleapis.com/user/metric_name"',
],
project_id=GCP_PROJECT_ID,
)
]
check = (
logging_log_metric_filter_and_alert_for_vpc_network_changes_enabled()
)
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert (
result[0].status_extended
== f"Log metric filter metric_name found with alert policy alert_policy associated in project {GCP_PROJECT_ID}."
)
assert result[0].resource_id == "metric_name"
assert result[0].resource_name == "metric_name"
assert result[0].project_id == GCP_PROJECT_ID
assert result[0].location == GCP_EU1_LOCATION
tests/providers/gcp/services/logging/logging_log_metric_filter_and_alert_for_vpc_network_route_changes_enabled/logging_log_metric_filter_and_alert_for_vpc_network_route_changes_enabled_test.py
+183
View File
@@ -0,0 +1,183 @@
from unittest.mock import MagicMock, patch
from tests.providers.gcp.gcp_fixtures import (
GCP_EU1_LOCATION,
GCP_PROJECT_ID,
set_mocked_gcp_provider,
)
class Test_logging_log_metric_filter_and_alert_for_vpc_network_route_changes_enabled:
def test_no_projects(self):
logging_client = MagicMock()
monitoring_client = MagicMock()
with patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_gcp_provider(),
), patch(
"prowler.providers.gcp.services.logging.logging_log_metric_filter_and_alert_for_vpc_network_route_changes_enabled.logging_log_metric_filter_and_alert_for_vpc_network_route_changes_enabled.logging_client",
new=logging_client,
), patch(
"prowler.providers.gcp.services.logging.logging_log_metric_filter_and_alert_for_vpc_network_route_changes_enabled.logging_log_metric_filter_and_alert_for_vpc_network_route_changes_enabled.monitoring_client",
new=monitoring_client,
):
from prowler.providers.gcp.services.logging.logging_log_metric_filter_and_alert_for_vpc_network_route_changes_enabled.logging_log_metric_filter_and_alert_for_vpc_network_route_changes_enabled import (
logging_log_metric_filter_and_alert_for_vpc_network_route_changes_enabled,
)
logging_client.metrics = []
logging_client.project_ids = []
monitoring_client.alert_policies = []
check = (
logging_log_metric_filter_and_alert_for_vpc_network_route_changes_enabled()
)
result = check.execute()
assert len(result) == 0
def test_no_log_metric_filters_no_alerts_one_project(self):
logging_client = MagicMock()
monitoring_client = MagicMock()
with patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_gcp_provider(),
), patch(
"prowler.providers.gcp.services.logging.logging_log_metric_filter_and_alert_for_vpc_network_route_changes_enabled.logging_log_metric_filter_and_alert_for_vpc_network_route_changes_enabled.logging_client",
new=logging_client,
), patch(
"prowler.providers.gcp.services.logging.logging_log_metric_filter_and_alert_for_vpc_network_route_changes_enabled.logging_log_metric_filter_and_alert_for_vpc_network_route_changes_enabled.monitoring_client",
new=monitoring_client,
):
from prowler.providers.gcp.services.logging.logging_log_metric_filter_and_alert_for_vpc_network_route_changes_enabled.logging_log_metric_filter_and_alert_for_vpc_network_route_changes_enabled import (
logging_log_metric_filter_and_alert_for_vpc_network_route_changes_enabled,
)
logging_client.metrics = []
logging_client.project_ids = [GCP_PROJECT_ID]
logging_client.region = GCP_EU1_LOCATION
monitoring_client.alert_policies = []
check = (
logging_log_metric_filter_and_alert_for_vpc_network_route_changes_enabled()
)
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== f"There are no log metric filters or alerts associated in project {GCP_PROJECT_ID}."
)
assert result[0].resource_id == GCP_PROJECT_ID
assert result[0].resource_name == ""
assert result[0].project_id == GCP_PROJECT_ID
assert result[0].location == GCP_EU1_LOCATION
def test_log_metric_filters_no_alerts(self):
logging_client = MagicMock()
monitoring_client = MagicMock()
with patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_gcp_provider(),
), patch(
"prowler.providers.gcp.services.logging.logging_log_metric_filter_and_alert_for_vpc_network_route_changes_enabled.logging_log_metric_filter_and_alert_for_vpc_network_route_changes_enabled.logging_client",
new=logging_client,
), patch(
"prowler.providers.gcp.services.logging.logging_log_metric_filter_and_alert_for_vpc_network_route_changes_enabled.logging_log_metric_filter_and_alert_for_vpc_network_route_changes_enabled.monitoring_client",
new=monitoring_client,
):
from prowler.providers.gcp.services.logging.logging_log_metric_filter_and_alert_for_vpc_network_route_changes_enabled.logging_log_metric_filter_and_alert_for_vpc_network_route_changes_enabled import (
logging_log_metric_filter_and_alert_for_vpc_network_route_changes_enabled,
)
from prowler.providers.gcp.services.logging.logging_service import Metric
logging_client.metrics = [
Metric(
name="metric_name",
type="custom.googleapis.com/invoice/paid/amount",
filter='resource.type="gce_route" AND (protoPayload.methodName:"compute.routes.delete" OR protoPayload.methodName:"compute.routes.insert")',
project_id=GCP_PROJECT_ID,
)
]
logging_client.project_ids = [GCP_PROJECT_ID]
logging_client.region = GCP_EU1_LOCATION
monitoring_client.alert_policies = []
check = (
logging_log_metric_filter_and_alert_for_vpc_network_route_changes_enabled()
)
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== f"Log metric filter metric_name found but no alerts associated in project {GCP_PROJECT_ID}."
)
assert result[0].resource_id == "metric_name"
assert result[0].resource_name == "metric_name"
assert result[0].project_id == GCP_PROJECT_ID
assert result[0].location == GCP_EU1_LOCATION
def test_log_metric_filters_with_alerts(self):
logging_client = MagicMock()
monitoring_client = MagicMock()
with patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_gcp_provider(),
), patch(
"prowler.providers.gcp.services.logging.logging_log_metric_filter_and_alert_for_vpc_network_route_changes_enabled.logging_log_metric_filter_and_alert_for_vpc_network_route_changes_enabled.logging_client",
new=logging_client,
), patch(
"prowler.providers.gcp.services.logging.logging_log_metric_filter_and_alert_for_vpc_network_route_changes_enabled.logging_log_metric_filter_and_alert_for_vpc_network_route_changes_enabled.monitoring_client",
new=monitoring_client,
):
from prowler.providers.gcp.services.logging.logging_log_metric_filter_and_alert_for_vpc_network_route_changes_enabled.logging_log_metric_filter_and_alert_for_vpc_network_route_changes_enabled import (
logging_log_metric_filter_and_alert_for_vpc_network_route_changes_enabled,
)
from prowler.providers.gcp.services.logging.logging_service import Metric
from prowler.providers.gcp.services.monitoring.monitoring_service import (
AlertPolicy,
)
logging_client.metrics = [
Metric(
name="metric_name",
type="metric_type",
filter='resource.type="gce_route" AND (protoPayload.methodName:"compute.routes.delete" OR protoPayload.methodName:"compute.routes.insert")',
project_id=GCP_PROJECT_ID,
)
]
logging_client.project_ids = [GCP_PROJECT_ID]
logging_client.region = GCP_EU1_LOCATION
monitoring_client.alert_policies = [
AlertPolicy(
name=f"projects/{GCP_PROJECT_ID}/alertPolicies/alert_policy",
display_name="alert_policy",
enabled=True,
filters=[
'metric.type = "logging.googleapis.com/user/metric_name"',
],
project_id=GCP_PROJECT_ID,
)
]
check = (
logging_log_metric_filter_and_alert_for_vpc_network_route_changes_enabled()
)
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert (
result[0].status_extended
== f"Log metric filter metric_name found with alert policy alert_policy associated in project {GCP_PROJECT_ID}."
)
assert result[0].resource_id == "metric_name"
assert result[0].resource_name == "metric_name"
assert result[0].project_id == GCP_PROJECT_ID
assert result[0].location == GCP_EU1_LOCATION
tests/providers/gcp/services/logging/logging_sink_created/logging_sink_created_test.py
+139
View File
@@ -0,0 +1,139 @@
from unittest.mock import MagicMock, patch
from tests.providers.gcp.gcp_fixtures import (
GCP_EU1_LOCATION,
GCP_PROJECT_ID,
set_mocked_gcp_provider,
)
class Test_logging_sink_created:
def test_no_projects(self):
logging_client = MagicMock()
with patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_gcp_provider(),
), patch(
"prowler.providers.gcp.services.logging.logging_sink_created.logging_sink_created.logging_client",
new=logging_client,
):
from prowler.providers.gcp.services.logging.logging_sink_created.logging_sink_created import (
logging_sink_created,
)
logging_client.project_ids = []
logging_client.sinks = []
check = logging_sink_created()
result = check.execute()
assert len(result) == 0
def test_no_sinks(self):
logging_client = MagicMock()
with patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_gcp_provider(),
), patch(
"prowler.providers.gcp.services.logging.logging_sink_created.logging_sink_created.logging_client",
new=logging_client,
):
from prowler.providers.gcp.services.logging.logging_sink_created.logging_sink_created import (
logging_sink_created,
)
logging_client.project_ids = [GCP_PROJECT_ID]
logging_client.region = GCP_EU1_LOCATION
logging_client.sinks = []
check = logging_sink_created()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== f"There are no logging sinks to export copies of all the log entries in project {GCP_PROJECT_ID}."
)
assert result[0].resource_id == GCP_PROJECT_ID
assert result[0].resource_name == ""
assert result[0].project_id == GCP_PROJECT_ID
assert result[0].location == GCP_EU1_LOCATION
def test_sink_all(self):
logging_client = MagicMock()
with patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_gcp_provider(),
), patch(
"prowler.providers.gcp.services.logging.logging_sink_created.logging_sink_created.logging_client",
new=logging_client,
):
from prowler.providers.gcp.services.logging.logging_service import Sink
from prowler.providers.gcp.services.logging.logging_sink_created.logging_sink_created import (
logging_sink_created,
)
logging_client.project_ids = [GCP_PROJECT_ID]
logging_client.region = GCP_EU1_LOCATION
logging_client.sinks = [
Sink(
name="sink1",
destination="destination1",
filter="all",
project_id=GCP_PROJECT_ID,
)
]
check = logging_sink_created()
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert (
result[0].status_extended
== f"Sink sink1 is enabled exporting copies of all the log entries in project {GCP_PROJECT_ID}."
)
assert result[0].resource_id == "sink1"
assert result[0].resource_name == "sink1"
assert result[0].project_id == GCP_PROJECT_ID
assert result[0].location == GCP_EU1_LOCATION
def test_sink_not_all(self):
logging_client = MagicMock()
with patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_gcp_provider(),
), patch(
"prowler.providers.gcp.services.logging.logging_sink_created.logging_sink_created.logging_client",
new=logging_client,
):
from prowler.providers.gcp.services.logging.logging_service import Sink
from prowler.providers.gcp.services.logging.logging_sink_created.logging_sink_created import (
logging_sink_created,
)
logging_client.project_ids = [GCP_PROJECT_ID]
logging_client.region = GCP_EU1_LOCATION
logging_client.sinks = [
Sink(
name="sink1",
destination="destination1",
filter="not all",
project_id=GCP_PROJECT_ID,
)
]
check = logging_sink_created()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== f"Sink sink1 is enabled but not exporting copies of all the log entries in project {GCP_PROJECT_ID}."
)
assert result[0].resource_id == "sink1"
assert result[0].resource_name == "sink1"
assert result[0].project_id == GCP_PROJECT_ID
assert result[0].location == GCP_EU1_LOCATION