fix: checks with no resource name (#8120)

This commit is contained in:
Daniel Barranquero
2025-06-27 11:40:43 +02:00
committed by GitHub
parent 7858c147f7
commit 7b58d1dd56
20 changed files with 1066 additions and 2 deletions
+1
View File
@@ -58,6 +58,7 @@ All notable changes to the **Prowler SDK** are documented in this file.
- Fix logic in VPC and ELBv2 checks [(#8077)](https://github.com/prowler-cloud/prowler/pull/8077)
- Retrieve correctly ECS Container insights settings [(#8097)](https://github.com/prowler-cloud/prowler/pull/8097)
- Fix correct handling for different accounts-dates in prowler dashboard compliance page [(#8108)](https://github.com/prowler-cloud/prowler/pull/8108)
- Handle empty name in Azure Defender and GCP checks [(#8120)](https://github.com/prowler-cloud/prowler/pull/8120)
---
+5 -1
View File
@@ -522,7 +522,11 @@ class Check_Report_GCP(Check_Report):
or getattr(resource, "name", None)
or ""
)
self.resource_name = resource_name or getattr(resource, "name", "")
self.resource_name = (
resource_name
or getattr(resource, "name", "")
or getattr(resource, "id", "")
)
self.project_id = project_id or getattr(resource, "project_id", "")
self.location = (
location
@@ -161,7 +161,8 @@ class Defender(AzureService):
{
security_contact_default.name: SecurityContacts(
resource_id=security_contact_default.id,
name=getattr(security_contact_default, "name", "default"),
name=getattr(security_contact_default, "name", "default")
or "default",
emails=security_contact_default.emails,
phone=security_contact_default.phone,
alert_notifications_minimal_severity=security_contact_default.alert_notifications.minimal_severity,
@@ -296,3 +296,49 @@ class Test_defender_additional_email_configured_with_a_security_contact:
result[0].resource_id
== f"/subscriptions/{AZURE_SUBSCRIPTION_ID}/providers/Microsoft.Security/securityContacts/default"
)
def test_defender_default_security_contact_not_found_empty_name(self):
resource_id = f"/subscriptions/{AZURE_SUBSCRIPTION_ID}/providers/Microsoft.Security/securityContacts/default"
defender_client = mock.MagicMock
defender_client.security_contacts = {
AZURE_SUBSCRIPTION_ID: {
resource_id: SecurityContacts(
resource_id=resource_id,
name="",
emails="",
phone="",
alert_notifications_minimal_severity="",
alert_notifications_state="",
notified_roles=[""],
notified_roles_state="",
)
}
}
contact = defender_client.security_contacts[AZURE_SUBSCRIPTION_ID][resource_id]
contact.name = getattr(contact, "name", "default") or "default"
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_azure_provider(),
),
mock.patch(
"prowler.providers.azure.services.defender.defender_additional_email_configured_with_a_security_contact.defender_additional_email_configured_with_a_security_contact.defender_client",
new=defender_client,
),
):
from prowler.providers.azure.services.defender.defender_additional_email_configured_with_a_security_contact.defender_additional_email_configured_with_a_security_contact import (
defender_additional_email_configured_with_a_security_contact,
)
check = defender_additional_email_configured_with_a_security_contact()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== f"There is not another correct email configured for subscription {AZURE_SUBSCRIPTION_ID}."
)
assert result[0].subscription == AZURE_SUBSCRIPTION_ID
assert result[0].resource_name == "default"
assert result[0].resource_id == resource_id
@@ -208,3 +208,50 @@ class Test_defender_ensure_notify_alerts_severity_is_high:
result[0].resource_id
== f"/subscriptions/{AZURE_SUBSCRIPTION_ID}/providers/Microsoft.Security/securityContacts/default"
)
def test_defender_default_security_contact_not_found_empty_name(self):
resource_id = f"/subscriptions/{AZURE_SUBSCRIPTION_ID}/providers/Microsoft.Security/securityContacts/default"
defender_client = mock.MagicMock
defender_client.security_contacts = {
AZURE_SUBSCRIPTION_ID: {
resource_id: SecurityContacts(
resource_id=resource_id,
name="",
emails="",
phone="",
alert_notifications_minimal_severity="",
alert_notifications_state="",
notified_roles=[""],
notified_roles_state="",
)
}
}
contact = defender_client.security_contacts[AZURE_SUBSCRIPTION_ID][resource_id]
contact.name = getattr(contact, "name", "default") or "default"
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_azure_provider(),
),
mock.patch(
"prowler.providers.azure.services.defender.defender_ensure_notify_alerts_severity_is_high.defender_ensure_notify_alerts_severity_is_high.defender_client",
new=defender_client,
),
):
from prowler.providers.azure.services.defender.defender_ensure_notify_alerts_severity_is_high.defender_ensure_notify_alerts_severity_is_high import (
defender_ensure_notify_alerts_severity_is_high,
)
check = defender_ensure_notify_alerts_severity_is_high()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== f"Notifications are not enabled for alerts with a minimum severity of high or lower in subscription {AZURE_SUBSCRIPTION_ID}."
)
assert result[0].subscription == AZURE_SUBSCRIPTION_ID
assert result[0].resource_name == "default"
assert result[0].resource_id == resource_id
@@ -208,3 +208,49 @@ class Test_defender_ensure_notify_emails_to_owners:
result[0].resource_id
== f"/subscriptions/{AZURE_SUBSCRIPTION_ID}/providers/Microsoft.Security/securityContacts/default"
)
def test_defender_default_security_contact_not_found_empty_name(self):
defender_client = mock.MagicMock()
resource_id = f"/subscriptions/{AZURE_SUBSCRIPTION_ID}/providers/Microsoft.Security/securityContacts/default"
defender_client.security_contacts = {
AZURE_SUBSCRIPTION_ID: {
resource_id: SecurityContacts(
resource_id=resource_id,
name="",
emails="",
phone="",
alert_notifications_minimal_severity="",
alert_notifications_state="",
notified_roles=[""],
notified_roles_state="",
)
}
}
contact = defender_client.security_contacts[AZURE_SUBSCRIPTION_ID][resource_id]
contact.name = getattr(contact, "name", "default") or "default"
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_azure_provider(),
),
mock.patch(
"prowler.providers.azure.services.defender.defender_ensure_notify_emails_to_owners.defender_ensure_notify_emails_to_owners.defender_client",
new=defender_client,
),
):
from prowler.providers.azure.services.defender.defender_ensure_notify_emails_to_owners.defender_ensure_notify_emails_to_owners import (
defender_ensure_notify_emails_to_owners,
)
check = defender_ensure_notify_emails_to_owners()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== f"The Owner role is not notified for subscription {AZURE_SUBSCRIPTION_ID}."
)
assert result[0].subscription == AZURE_SUBSCRIPTION_ID
assert result[0].resource_name == "default"
assert result[0].resource_id == resource_id
@@ -128,3 +128,103 @@ class Test_compute_project_os_login_enabled:
assert result[0].resource_name == "test"
assert result[0].location == "global"
assert result[0].project_id == GCP_PROJECT_ID
def test_one_compliant_project_empty_project_name(self):
from prowler.providers.gcp.services.compute.compute_service import Project
project = Project(
id=GCP_PROJECT_ID,
enable_oslogin=True,
)
compute_client = mock.MagicMock()
compute_client.project_ids = [GCP_PROJECT_ID]
compute_client.compute_projects = [project]
compute_client.projects = {
GCP_PROJECT_ID: GCPProject(
id=GCP_PROJECT_ID,
number="123456789012",
name="",
labels={},
lifecycle_state="ACTIVE",
)
}
compute_client.region = "global"
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_gcp_provider(),
),
mock.patch(
"prowler.providers.gcp.services.compute.compute_project_os_login_enabled.compute_project_os_login_enabled.compute_client",
new=compute_client,
),
):
from prowler.providers.gcp.services.compute.compute_project_os_login_enabled.compute_project_os_login_enabled import (
compute_project_os_login_enabled,
)
check = compute_project_os_login_enabled()
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert search(
f"Project {project.id} has OS Login enabled",
result[0].status_extended,
)
assert result[0].resource_id == project.id
assert result[0].resource_name == GCP_PROJECT_ID
assert result[0].location == "global"
assert result[0].project_id == GCP_PROJECT_ID
def test_one_non_compliant_project_empty_project_name(self):
from prowler.providers.gcp.services.compute.compute_service import Project
project = Project(
id=GCP_PROJECT_ID,
enable_oslogin=False,
)
compute_client = mock.MagicMock()
compute_client.project_ids = [GCP_PROJECT_ID]
compute_client.compute_projects = [project]
compute_client.projects = {
GCP_PROJECT_ID: GCPProject(
id=GCP_PROJECT_ID,
number="123456789012",
name="",
labels={},
lifecycle_state="ACTIVE",
)
}
compute_client.region = "global"
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_gcp_provider(),
),
mock.patch(
"prowler.providers.gcp.services.compute.compute_project_os_login_enabled.compute_project_os_login_enabled.compute_client",
new=compute_client,
),
):
from prowler.providers.gcp.services.compute.compute_project_os_login_enabled.compute_project_os_login_enabled import (
compute_project_os_login_enabled,
)
check = compute_project_os_login_enabled()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert search(
f"Project {project.id} does not have OS Login enabled",
result[0].status_extended,
)
assert result[0].resource_id == project.id
assert result[0].resource_name == GCP_PROJECT_ID
assert result[0].location == "global"
assert result[0].project_id == GCP_PROJECT_ID
@@ -129,3 +129,103 @@ class Test_iam_audit_logs_enabled:
assert r.resource_name == "test"
assert r.project_id == GCP_PROJECT_ID
assert r.location == cloudresourcemanager_client.region
def test_compliant_project_empty_project_name(self):
from prowler.providers.gcp.services.cloudresourcemanager.cloudresourcemanager_service import (
Project,
)
project1 = Project(id=GCP_PROJECT_ID, audit_logging=True)
cloudresourcemanager_client = mock.MagicMock()
cloudresourcemanager_client.project_ids = [GCP_PROJECT_ID]
cloudresourcemanager_client.cloud_resource_manager_projects = [project1]
cloudresourcemanager_client.projects = {
GCP_PROJECT_ID: GCPProject(
id=GCP_PROJECT_ID,
number="123456789012",
name="",
labels={},
lifecycle_state="ACTIVE",
)
}
cloudresourcemanager_client.region = "global"
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_gcp_provider(),
),
mock.patch(
"prowler.providers.gcp.services.iam.iam_audit_logs_enabled.iam_audit_logs_enabled.cloudresourcemanager_client",
new=cloudresourcemanager_client,
),
):
from prowler.providers.gcp.services.iam.iam_audit_logs_enabled.iam_audit_logs_enabled import (
iam_audit_logs_enabled,
)
check = iam_audit_logs_enabled()
result = check.execute()
assert len(result) == 1
for idx, r in enumerate(result):
assert r.status == "PASS"
assert search(
"Audit Logs are enabled for project",
r.status_extended,
)
assert r.resource_id == GCP_PROJECT_ID
assert r.resource_name == GCP_PROJECT_ID
assert r.project_id == GCP_PROJECT_ID
assert r.location == cloudresourcemanager_client.region
def test_uncompliant_project_empty_project_name(self):
from prowler.providers.gcp.services.cloudresourcemanager.cloudresourcemanager_service import (
Project,
)
project1 = Project(id=GCP_PROJECT_ID, audit_logging=False)
cloudresourcemanager_client = mock.MagicMock()
cloudresourcemanager_client.project_ids = [GCP_PROJECT_ID]
cloudresourcemanager_client.cloud_resource_manager_projects = [project1]
cloudresourcemanager_client.projects = {
GCP_PROJECT_ID: GCPProject(
id=GCP_PROJECT_ID,
number="123456789012",
name="",
labels={},
lifecycle_state="ACTIVE",
)
}
cloudresourcemanager_client.region = "global"
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_gcp_provider(),
),
mock.patch(
"prowler.providers.gcp.services.iam.iam_audit_logs_enabled.iam_audit_logs_enabled.cloudresourcemanager_client",
new=cloudresourcemanager_client,
),
):
from prowler.providers.gcp.services.iam.iam_audit_logs_enabled.iam_audit_logs_enabled import (
iam_audit_logs_enabled,
)
check = iam_audit_logs_enabled()
result = check.execute()
assert len(result) == 1
for idx, r in enumerate(result):
assert r.status == "FAIL"
assert search(
"Audit Logs are not enabled for project",
r.status_extended,
)
assert r.resource_id == GCP_PROJECT_ID
assert r.resource_name == GCP_PROJECT_ID
assert r.project_id == GCP_PROJECT_ID
assert r.location == cloudresourcemanager_client.region
@@ -212,3 +212,49 @@ class Test_iam_no_service_roles_at_project_level:
assert result[0].resource_name == binding.role
assert result[0].project_id == GCP_PROJECT_ID
assert result[0].location == cloudresourcemanager_client.region
def test_iam_no_bindings_empty_project_name(self):
cloudresourcemanager_client = mock.MagicMock()
cloudresourcemanager_client.bindings = []
cloudresourcemanager_client.project_ids = [GCP_PROJECT_ID]
cloudresourcemanager_client.region = "global"
cloudresourcemanager_client.projects = {
GCP_PROJECT_ID: GCPProject(
id=GCP_PROJECT_ID,
number="123456789012",
name="",
labels={},
lifecycle_state="ACTIVE",
)
}
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_gcp_provider(),
),
mock.patch(
"prowler.providers.gcp.services.cloudresourcemanager.cloudresourcemanager_service.CloudResourceManager",
new=cloudresourcemanager_client,
),
mock.patch(
"prowler.providers.gcp.services.iam.iam_no_service_roles_at_project_level.iam_no_service_roles_at_project_level.cloudresourcemanager_client",
new=cloudresourcemanager_client,
),
):
from prowler.providers.gcp.services.iam.iam_no_service_roles_at_project_level.iam_no_service_roles_at_project_level import (
iam_no_service_roles_at_project_level,
)
check = iam_no_service_roles_at_project_level()
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert search(
"No IAM Users assigned to service roles at project level",
result[0].status_extended,
)
assert result[0].resource_id == GCP_PROJECT_ID
assert result[0].resource_name == GCP_PROJECT_ID
assert result[0].project_id == GCP_PROJECT_ID
assert result[0].location == cloudresourcemanager_client.region
@@ -173,3 +173,110 @@ class Test_iam_role_kms_enforce_separation_of_duties:
assert r.resource_id == GCP_PROJECT_ID
assert r.project_id == GCP_PROJECT_ID
assert r.location == cloudresourcemanager_client.region
def test_iam_no_bindings_empty_project_name(self):
cloudresourcemanager_client = mock.MagicMock()
cloudresourcemanager_client.bindings = []
cloudresourcemanager_client.project_ids = [GCP_PROJECT_ID]
cloudresourcemanager_client.region = "global"
cloudresourcemanager_client.projects = {
GCP_PROJECT_ID: GCPProject(
id=GCP_PROJECT_ID,
number="123456789012",
name="",
labels={},
lifecycle_state="ACTIVE",
)
}
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_gcp_provider(),
),
mock.patch(
"prowler.providers.gcp.services.iam.iam_role_kms_enforce_separation_of_duties.iam_role_kms_enforce_separation_of_duties.cloudresourcemanager_client",
new=cloudresourcemanager_client,
),
):
from prowler.providers.gcp.services.iam.iam_role_kms_enforce_separation_of_duties.iam_role_kms_enforce_separation_of_duties import (
iam_role_kms_enforce_separation_of_duties,
)
check = iam_role_kms_enforce_separation_of_duties()
result = check.execute()
assert len(result) == 1
for idx, r in enumerate(result):
assert r.status == "PASS"
assert search(
"Principle of separation of duties was enforced for KMS-Related Roles",
r.status_extended,
)
assert r.resource_id == GCP_PROJECT_ID
assert r.resource_name == GCP_PROJECT_ID
assert r.project_id == GCP_PROJECT_ID
assert r.location == cloudresourcemanager_client.region
def test_uncompliant_binding_empty_project_name(self):
from prowler.providers.gcp.services.cloudresourcemanager.cloudresourcemanager_service import (
Binding,
)
binding1 = Binding(
role="roles/cloudkms.admin",
members=["serviceAccount:685829395199@cloudbuild.gserviceaccount.com"],
project_id=GCP_PROJECT_ID,
)
binding2 = Binding(
role="roles/cloudkms.cryptoKeyEncrypterDecrypter",
members=["serviceAccount:685829395199@cloudbuild.gserviceaccount.com"],
project_id=GCP_PROJECT_ID,
)
binding3 = Binding(
role="roles/connectors.managedZoneViewer",
members=["serviceAccount:685829395199@cloudbuild.gserviceaccount.com"],
project_id=GCP_PROJECT_ID,
)
cloudresourcemanager_client = mock.MagicMock()
cloudresourcemanager_client.project_ids = [GCP_PROJECT_ID]
cloudresourcemanager_client.bindings = [binding1, binding2, binding3]
cloudresourcemanager_client.region = "global"
cloudresourcemanager_client.projects = {
GCP_PROJECT_ID: GCPProject(
id=GCP_PROJECT_ID,
number="123456789012",
name="",
labels={},
lifecycle_state="ACTIVE",
)
}
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_gcp_provider(),
),
mock.patch(
"prowler.providers.gcp.services.iam.iam_role_kms_enforce_separation_of_duties.iam_role_kms_enforce_separation_of_duties.cloudresourcemanager_client",
new=cloudresourcemanager_client,
),
):
from prowler.providers.gcp.services.iam.iam_role_kms_enforce_separation_of_duties.iam_role_kms_enforce_separation_of_duties import (
iam_role_kms_enforce_separation_of_duties,
)
check = iam_role_kms_enforce_separation_of_duties()
result = check.execute()
assert len(result) == 1
for idx, r in enumerate(result):
assert r.status == "FAIL"
assert search(
"Principle of separation of duties was not enforced for KMS-Related Roles",
r.status_extended,
)
assert r.resource_id == GCP_PROJECT_ID
assert r.resource_name == GCP_PROJECT_ID
assert r.project_id == GCP_PROJECT_ID
assert r.location == cloudresourcemanager_client.region
@@ -173,3 +173,110 @@ class Test_iam_role_sa_enforce_separation_of_duties:
assert r.resource_id == GCP_PROJECT_ID
assert r.project_id == GCP_PROJECT_ID
assert r.location == cloudresourcemanager_client.region
def test_iam_no_bindings_empty_project_name(self):
cloudresourcemanager_client = mock.MagicMock()
cloudresourcemanager_client.bindings = []
cloudresourcemanager_client.project_ids = [GCP_PROJECT_ID]
cloudresourcemanager_client.region = "global"
cloudresourcemanager_client.projects = {
GCP_PROJECT_ID: GCPProject(
id=GCP_PROJECT_ID,
number="123456789012",
name="",
labels={},
lifecycle_state="ACTIVE",
)
}
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_gcp_provider(),
),
mock.patch(
"prowler.providers.gcp.services.iam.iam_role_sa_enforce_separation_of_duties.iam_role_sa_enforce_separation_of_duties.cloudresourcemanager_client",
new=cloudresourcemanager_client,
),
):
from prowler.providers.gcp.services.iam.iam_role_sa_enforce_separation_of_duties.iam_role_sa_enforce_separation_of_duties import (
iam_role_sa_enforce_separation_of_duties,
)
check = iam_role_sa_enforce_separation_of_duties()
result = check.execute()
assert len(result) == 1
for idx, r in enumerate(result):
assert r.status == "PASS"
assert search(
"Principle of separation of duties was enforced for Service-Account Related Roles",
r.status_extended,
)
assert r.resource_id == GCP_PROJECT_ID
assert r.resource_name == GCP_PROJECT_ID
assert r.project_id == GCP_PROJECT_ID
assert r.location == cloudresourcemanager_client.region
def test_one_uncompliant_binding_empty_project_name(self):
from prowler.providers.gcp.services.cloudresourcemanager.cloudresourcemanager_service import (
Binding,
)
binding1 = Binding(
role="roles/iam.serviceAccountUser",
members=["serviceAccount:685829395199@cloudbuild.gserviceaccount.com"],
project_id=GCP_PROJECT_ID,
)
binding2 = Binding(
role="roles/compute.serviceAgent",
members=["serviceAccount:685829395199@cloudbuild.gserviceaccount.com"],
project_id=GCP_PROJECT_ID,
)
binding3 = Binding(
role="roles/connectors.managedZoneViewer",
members=["serviceAccount:685829395199@cloudbuild.gserviceaccount.com"],
project_id=GCP_PROJECT_ID,
)
cloudresourcemanager_client = mock.MagicMock()
cloudresourcemanager_client.project_ids = [GCP_PROJECT_ID]
cloudresourcemanager_client.bindings = [binding1, binding2, binding3]
cloudresourcemanager_client.region = "global"
cloudresourcemanager_client.projects = {
GCP_PROJECT_ID: GCPProject(
id=GCP_PROJECT_ID,
number="123456789012",
name="",
labels={},
lifecycle_state="ACTIVE",
)
}
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_gcp_provider(),
),
mock.patch(
"prowler.providers.gcp.services.iam.iam_role_sa_enforce_separation_of_duties.iam_role_sa_enforce_separation_of_duties.cloudresourcemanager_client",
new=cloudresourcemanager_client,
),
):
from prowler.providers.gcp.services.iam.iam_role_sa_enforce_separation_of_duties.iam_role_sa_enforce_separation_of_duties import (
iam_role_sa_enforce_separation_of_duties,
)
check = iam_role_sa_enforce_separation_of_duties()
result = check.execute()
assert len(result) == 1
for idx, r in enumerate(result):
assert r.status == "FAIL"
assert search(
"Principle of separation of duties was not enforced for Service-Account Related Roles",
r.status_extended,
)
assert r.resource_id == GCP_PROJECT_ID
assert r.resource_name == GCP_PROJECT_ID
assert r.project_id == GCP_PROJECT_ID
assert r.location == cloudresourcemanager_client.region
@@ -93,6 +93,58 @@ class Test_logging_log_metric_filter_and_alert_for_audit_configuration_changes_e
assert result[0].project_id == GCP_PROJECT_ID
assert result[0].location == GCP_EU1_LOCATION
def test_no_log_metric_filters_no_alerts_one_project_empty_name(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
logging_client.projects = {
GCP_PROJECT_ID: GCPProject(
id=GCP_PROJECT_ID,
number="123456789012",
name="",
labels={},
lifecycle_state="ACTIVE",
)
}
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 == GCP_PROJECT_ID
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()
@@ -93,6 +93,58 @@ class Test_logging_log_metric_filter_and_alert_for_bucket_permission_changes_ena
assert result[0].project_id == GCP_PROJECT_ID
assert result[0].location == GCP_EU1_LOCATION
def test_no_log_metric_filters_no_alerts_one_project_empty_name(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
logging_client.projects = {
GCP_PROJECT_ID: GCPProject(
id=GCP_PROJECT_ID,
number="123456789012",
name="",
labels={},
lifecycle_state="ACTIVE",
)
}
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 == GCP_PROJECT_ID
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()
@@ -93,6 +93,58 @@ class Test_logging_log_metric_filter_and_alert_for_custom_role_changes_enabled:
assert result[0].project_id == GCP_PROJECT_ID
assert result[0].location == GCP_EU1_LOCATION
def test_no_log_metric_filters_no_alerts_one_project_empty_name(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
logging_client.projects = {
GCP_PROJECT_ID: GCPProject(
id=GCP_PROJECT_ID,
number="123456789012",
name="",
labels={},
lifecycle_state="ACTIVE",
)
}
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 == GCP_PROJECT_ID
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()
@@ -93,6 +93,58 @@ class Test_logging_log_metric_filter_and_alert_for_project_ownership_changes_ena
assert result[0].project_id == GCP_PROJECT_ID
assert result[0].location == GCP_EU1_LOCATION
def test_no_log_metric_filters_no_alerts_one_project_empty_name(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
logging_client.projects = {
GCP_PROJECT_ID: GCPProject(
id=GCP_PROJECT_ID,
number="123456789012",
name="",
labels={},
lifecycle_state="ACTIVE",
)
}
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 == GCP_PROJECT_ID
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()
@@ -93,6 +93,58 @@ class Test_logging_log_metric_filter_and_alert_for_sql_instance_configuration_ch
assert result[0].project_id == GCP_PROJECT_ID
assert result[0].location == GCP_EU1_LOCATION
def test_no_log_metric_filters_no_alerts_one_project_empty_name(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
logging_client.projects = {
GCP_PROJECT_ID: GCPProject(
id=GCP_PROJECT_ID,
number="123456789012",
name="",
labels={},
lifecycle_state="ACTIVE",
)
}
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 == GCP_PROJECT_ID
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()
@@ -93,6 +93,58 @@ class Test_logging_log_metric_filter_and_alert_for_vpc_firewall_rule_changes_ena
assert result[0].project_id == GCP_PROJECT_ID
assert result[0].location == GCP_EU1_LOCATION
def test_no_log_metric_filters_no_alerts_one_project_empty_name(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
logging_client.projects = {
GCP_PROJECT_ID: GCPProject(
id=GCP_PROJECT_ID,
number="123456789012",
name="",
labels={},
lifecycle_state="ACTIVE",
)
}
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 == GCP_PROJECT_ID
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()
@@ -93,6 +93,58 @@ class Test_logging_log_metric_filter_and_alert_for_vpc_network_changes_enabled:
assert result[0].project_id == GCP_PROJECT_ID
assert result[0].location == GCP_EU1_LOCATION
def test_no_log_metric_filters_no_alerts_one_project_empty_name(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
logging_client.projects = {
GCP_PROJECT_ID: GCPProject(
id=GCP_PROJECT_ID,
number="123456789012",
name="",
labels={},
lifecycle_state="ACTIVE",
)
}
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 == GCP_PROJECT_ID
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()
@@ -93,6 +93,58 @@ class Test_logging_log_metric_filter_and_alert_for_vpc_network_route_changes_ena
assert result[0].project_id == GCP_PROJECT_ID
assert result[0].location == GCP_EU1_LOCATION
def test_no_log_metric_filters_no_alerts_one_project_empty_name(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
logging_client.projects = {
GCP_PROJECT_ID: GCPProject(
id=GCP_PROJECT_ID,
number="123456789012",
name="",
labels={},
lifecycle_state="ACTIVE",
)
}
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 == GCP_PROJECT_ID
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()
@@ -168,3 +168,46 @@ class Test_logging_sink_created:
assert result[0].resource_name == "test"
assert result[0].project_id == GCP_PROJECT_ID
assert result[0].location == GCP_EU1_LOCATION
def test_no_sinks_empty_project_name(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 = []
logging_client.projects = {
GCP_PROJECT_ID: GCPProject(
id=GCP_PROJECT_ID,
number="123456789012",
name="",
labels={},
lifecycle_state="ACTIVE",
)
}
check = logging_sink_created()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert result[0].resource_id == GCP_PROJECT_ID
assert result[0].resource_name == GCP_PROJECT_ID
assert result[0].project_id == GCP_PROJECT_ID
assert result[0].location == GCP_EU1_LOCATION
assert (
result[0].status_extended
== f"There are no logging sinks to export copies of all the log entries in project {GCP_PROJECT_ID}."
)