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

feat(gcp): add resource metadata to report (#6500)

Browse Source 
Co-authored-by: MrCloudSec <hello@mistercloudsec.com>
This commit is contained in:
Pedro Martín
2025-01-16 00:09:35 +01:00
committed by Pepe Fagoaga
parent 4c6be5e283
commit 32253ca4f7
160 changed files with 1249 additions and 895 deletions
Download Patch File Download Diff File Expand all files Collapse all files
prowler/lib/check/models.py
+23 -6
View File
@@ -513,12 +513,29 @@ class Check_Report_GCP(Check_Report):
project_id: str
location: str
def __init__(self, metadata):
super().__init__(metadata)
self.resource_name = ""
self.resource_id = ""
self.project_id = ""
self.location = ""
def __init__(
self,
metadata,
resource_metadata,
location=None,
resource_name=None,
resource_id=None,
project_id=None,
):
super().__init__(metadata, resource_metadata)
self.resource_id = (
resource_id
or getattr(resource_metadata, "id", None)
or getattr(resource_metadata, "name", None)
or ""
)
self.resource_name = resource_name or getattr(resource_metadata, "name", "")
self.project_id = project_id or getattr(resource_metadata, "project_id", "")
self.location = (
location
or getattr(resource_metadata, "location", "")
or getattr(resource_metadata, "region", "")
)
@dataclass
prowler/providers/gcp/lib/service/service.py
+1
View File
@@ -30,6 +30,7 @@ class GCPService:
)
# Only project ids that have their API enabled will be scanned
self.project_ids = self.__is_api_active__(provider.project_ids)
self.projects = provider.projects
self.default_project_id = provider.default_project_id
self.audit_config = provider.audit_config
self.fixer_config = provider.fixer_config
prowler/providers/gcp/services/apikeys/apikeys_api_restrictions_configured/apikeys_api_restrictions_configured.py
+5 -5
View File
@@ -6,12 +6,12 @@ class apikeys_api_restrictions_configured(Check):
def execute(self) -> Check_Report_GCP:
findings = []
for key in apikeys_client.keys:
report = Check_Report_GCP(self.metadata())
report.project_id = key.project_id
report.resource_id = key.id
report.resource_name = key.name
report = Check_Report_GCP(
metadata=self.metadata(),
resource_metadata=key,
location=apikeys_client.region,
)
report.status = "PASS"
report.location = apikeys_client.region
report.status_extended = f"API key {key.name} has restrictions configured."
if key.restrictions == {} or any(
[
prowler/providers/gcp/services/apikeys/apikeys_key_exists/apikeys_key_exists.py
+6 -4
View File
@@ -6,10 +6,12 @@ class apikeys_key_exists(Check):
def execute(self) -> Check_Report_GCP:
findings = []
for project in apikeys_client.project_ids:
report = Check_Report_GCP(self.metadata())
report.project_id = project
report.resource_id = project
report.location = apikeys_client.region
report = Check_Report_GCP(
metadata=self.metadata(),
resource_metadata=apikeys_client.projects[project],
project_id=project,
location=apikeys_client.region,
)
report.status = "PASS"
report.status_extended = f"Project {project} does not have active API Keys."
for key in apikeys_client.keys:
prowler/providers/gcp/services/apikeys/apikeys_key_rotated_in_90_days/apikeys_key_rotated_in_90_days.py
+5 -5
View File
@@ -8,11 +8,11 @@ class apikeys_key_rotated_in_90_days(Check):
def execute(self) -> Check_Report_GCP:
findings = []
for key in apikeys_client.keys:
report = Check_Report_GCP(self.metadata())
report.project_id = key.project_id
report.resource_id = key.id
report.resource_name = key.name
report.location = apikeys_client.region
report = Check_Report_GCP(
metadata=self.metadata(),
resource_metadata=key,
location=apikeys_client.region,
)
report.status = "PASS"
report.status_extended = f"API key {key.name} created in less than 90 days."
if (
prowler/providers/gcp/services/artifacts/artifacts_container_analysis_enabled/artifacts_container_analysis_enabled.py
+8 -5
View File
@@ -8,11 +8,14 @@ class artifacts_container_analysis_enabled(Check):
def execute(self) -> Check_Report_GCP:
findings = []
for project_id in serviceusage_client.project_ids:
report = Check_Report_GCP(self.metadata())
report.project_id = project_id
report.resource_id = "containeranalysis.googleapis.com"
report.resource_name = "AR Container Analysis"
report.location = serviceusage_client.region
report = Check_Report_GCP(
metadata=self.metadata(),
resource_metadata=serviceusage_client.projects[project_id],
resource_id="containeranalysis.googleapis.com",
resource_name="AR Container Analysis",
project_id=project_id,
location=serviceusage_client.region,
)
report.status = "FAIL"
report.status_extended = (
f"AR Container Analysis is not enabled in project {project_id}."
prowler/providers/gcp/services/bigquery/bigquery_dataset_cmk_encryption/bigquery_dataset_cmk_encryption.py
+3 -5
View File
@@ -6,11 +6,9 @@ class bigquery_dataset_cmk_encryption(Check):
def execute(self) -> Check_Report_GCP:
findings = []
for dataset in bigquery_client.datasets:
report = Check_Report_GCP(self.metadata())
report.project_id = dataset.project_id
report.resource_id = dataset.id
report.resource_name = dataset.name
report.location = dataset.region
report = Check_Report_GCP(
metadata=self.metadata(), resource_metadata=dataset
)
report.status = "PASS"
report.status_extended = f"Dataset {dataset.name} is encrypted with Customer-Managed Keys (CMKs)."
if not dataset.cmk_encryption:
prowler/providers/gcp/services/bigquery/bigquery_dataset_public_access/bigquery_dataset_public_access.py
+3 -5
View File
@@ -6,11 +6,9 @@ class bigquery_dataset_public_access(Check):
def execute(self) -> Check_Report_GCP:
findings = []
for dataset in bigquery_client.datasets:
report = Check_Report_GCP(self.metadata())
report.project_id = dataset.project_id
report.resource_id = dataset.id
report.resource_name = dataset.name
report.location = dataset.region
report = Check_Report_GCP(
metadata=self.metadata(), resource_metadata=dataset
)
report.status = "PASS"
report.status_extended = (
f"Dataset {dataset.name} is not publicly accessible."
prowler/providers/gcp/services/bigquery/bigquery_table_cmk_encryption/bigquery_table_cmk_encryption.py
+1 -5
View File
@@ -6,11 +6,7 @@ class bigquery_table_cmk_encryption(Check):
def execute(self) -> Check_Report_GCP:
findings = []
for table in bigquery_client.tables:
report = Check_Report_GCP(self.metadata())
report.project_id = table.project_id
report.resource_id = table.id
report.resource_name = table.name
report.location = table.region
report = Check_Report_GCP(metadata=self.metadata(), resource_metadata=table)
report.status = "PASS"
report.status_extended = (
f"Table {table.name} is encrypted with Customer-Managed Keys (CMKs)."
prowler/providers/gcp/services/cloudresourcemanager/cloudresourcemanager_service.py
+2 -2
View File
@@ -10,7 +10,7 @@ class CloudResourceManager(GCPService):
super().__init__(__class__.__name__, provider)
self.bindings = []
self.projects = []
self.cloud_resource_manager_projects = []
self.organizations = []
self._get_iam_policy()
self._get_organizations()
@@ -24,7 +24,7 @@ class CloudResourceManager(GCPService):
audit_logging = False
if policy.get("auditConfigs"):
audit_logging = True
self.projects.append(
self.cloud_resource_manager_projects.append(
Project(id=project_id, audit_logging=audit_logging)
)
for binding in policy["bindings"]:
prowler/providers/gcp/services/cloudsql/cloudsql_instance_automated_backups/cloudsql_instance_automated_backups.py
+3 -5
View File
@@ -6,11 +6,9 @@ class cloudsql_instance_automated_backups(Check):
def execute(self) -> Check_Report_GCP:
findings = []
for instance in cloudsql_client.instances:
report = Check_Report_GCP(self.metadata())
report.project_id = instance.project_id
report.resource_id = instance.name
report.resource_name = instance.name
report.location = instance.region
report = Check_Report_GCP(
metadata=self.metadata(), resource_metadata=instance
)
report.status = "PASS"
report.status_extended = (
f"Database Instance {instance.name} has automated backups configured."
prowler/providers/gcp/services/cloudsql/cloudsql_instance_mysql_local_infile_flag/cloudsql_instance_mysql_local_infile_flag.py
+3 -5
View File
@@ -7,11 +7,9 @@ class cloudsql_instance_mysql_local_infile_flag(Check):
findings = []
for instance in cloudsql_client.instances:
if "MYSQL" in instance.version:
report = Check_Report_GCP(self.metadata())
report.project_id = instance.project_id
report.resource_id = instance.name
report.resource_name = instance.name
report.location = instance.region
report = Check_Report_GCP(
metadata=self.metadata(), resource_metadata=instance
)
report.status = "FAIL"
report.status_extended = f"MySQL Instance {instance.name} does not have 'local_infile' flag set to 'off'."
for flag in instance.flags:
prowler/providers/gcp/services/cloudsql/cloudsql_instance_mysql_skip_show_database_flag/cloudsql_instance_mysql_skip_show_database_flag.py
+3 -5
View File
@@ -7,11 +7,9 @@ class cloudsql_instance_mysql_skip_show_database_flag(Check):
findings = []
for instance in cloudsql_client.instances:
if "MYSQL" in instance.version:
report = Check_Report_GCP(self.metadata())
report.project_id = instance.project_id
report.resource_id = instance.name
report.resource_name = instance.name
report.location = instance.region
report = Check_Report_GCP(
metadata=self.metadata(), resource_metadata=instance
)
report.status = "FAIL"
report.status_extended = f"MySQL Instance {instance.name} does not have 'skip_show_database' flag set to 'on'."
for flag in instance.flags:
prowler/providers/gcp/services/cloudsql/cloudsql_instance_postgres_enable_pgaudit_flag/cloudsql_instance_postgres_enable_pgaudit_flag.py
+3 -5
View File
@@ -7,11 +7,9 @@ class cloudsql_instance_postgres_enable_pgaudit_flag(Check):
findings = []
for instance in cloudsql_client.instances:
if "POSTGRES" in instance.version:
report = Check_Report_GCP(self.metadata())
report.project_id = instance.project_id
report.resource_id = instance.name
report.resource_name = instance.name
report.location = instance.region
report = Check_Report_GCP(
metadata=self.metadata(), resource_metadata=instance
)
report.status = "FAIL"
report.status_extended = f"PostgreSQL Instance {instance.name} does not have 'cloudsql.enable_pgaudit' flag set to 'on'."
for flag in instance.flags:
prowler/providers/gcp/services/cloudsql/cloudsql_instance_postgres_log_connections_flag/cloudsql_instance_postgres_log_connections_flag.py
+3 -5
View File
@@ -7,11 +7,9 @@ class cloudsql_instance_postgres_log_connections_flag(Check):
findings = []
for instance in cloudsql_client.instances:
if "POSTGRES" in instance.version:
report = Check_Report_GCP(self.metadata())
report.project_id = instance.project_id
report.resource_id = instance.name
report.resource_name = instance.name
report.location = instance.region
report = Check_Report_GCP(
metadata=self.metadata(), resource_metadata=instance
)
report.status = "FAIL"
report.status_extended = f"PostgreSQL Instance {instance.name} does not have 'log_connections' flag set to 'on'."
for flag in instance.flags:
prowler/providers/gcp/services/cloudsql/cloudsql_instance_postgres_log_disconnections_flag/cloudsql_instance_postgres_log_disconnections_flag.py
+3 -5
View File
@@ -7,11 +7,9 @@ class cloudsql_instance_postgres_log_disconnections_flag(Check):
findings = []
for instance in cloudsql_client.instances:
if "POSTGRES" in instance.version:
report = Check_Report_GCP(self.metadata())
report.project_id = instance.project_id
report.resource_id = instance.name
report.resource_name = instance.name
report.location = instance.region
report = Check_Report_GCP(
metadata=self.metadata(), resource_metadata=instance
)
report.status = "FAIL"
report.status_extended = f"PostgreSQL Instance {instance.name} does not have 'log_disconnections' flag set to 'on'."
for flag in instance.flags:
prowler/providers/gcp/services/cloudsql/cloudsql_instance_postgres_log_error_verbosity_flag/cloudsql_instance_postgres_log_error_verbosity_flag.py
+3 -5
View File
@@ -7,11 +7,9 @@ class cloudsql_instance_postgres_log_error_verbosity_flag(Check):
findings = []
for instance in cloudsql_client.instances:
if "POSTGRES" in instance.version:
report = Check_Report_GCP(self.metadata())
report.project_id = instance.project_id
report.resource_id = instance.name
report.resource_name = instance.name
report.location = instance.region
report = Check_Report_GCP(
metadata=self.metadata(), resource_metadata=instance
)
report.status = "PASS"
report.status_extended = f"PostgreSQL Instance {instance.name} has 'log_error_verbosity' flag set to 'default'."
for flag in instance.flags:
prowler/providers/gcp/services/cloudsql/cloudsql_instance_postgres_log_min_duration_statement_flag/cloudsql_instance_postgres_log_min_duration_statement_flag.py
+3 -5
View File
@@ -7,11 +7,9 @@ class cloudsql_instance_postgres_log_min_duration_statement_flag(Check):
findings = []
for instance in cloudsql_client.instances:
if "POSTGRES" in instance.version:
report = Check_Report_GCP(self.metadata())
report.project_id = instance.project_id
report.resource_id = instance.name
report.resource_name = instance.name
report.location = instance.region
report = Check_Report_GCP(
metadata=self.metadata(), resource_metadata=instance
)
report.status = "PASS"
report.status_extended = f"PostgreSQL Instance {instance.name} has 'log_min_duration_statement' flag set to '-1'."
for flag in instance.flags:
prowler/providers/gcp/services/cloudsql/cloudsql_instance_postgres_log_min_error_statement_flag/cloudsql_instance_postgres_log_min_error_statement_flag.py
+3 -5
View File
@@ -8,11 +8,9 @@ class cloudsql_instance_postgres_log_min_error_statement_flag(Check):
findings = []
for instance in cloudsql_client.instances:
if "POSTGRES" in instance.version:
report = Check_Report_GCP(self.metadata())
report.project_id = instance.project_id
report.resource_id = instance.name
report.resource_name = instance.name
report.location = instance.region
report = Check_Report_GCP(
metadata=self.metadata(), resource_metadata=instance
)
report.status = "PASS"
report.status_extended = f"PostgreSQL Instance {instance.name} has 'log_min_error_statement' flag set minimum to '{desired_log_min_error_statement}'."
prowler/providers/gcp/services/cloudsql/cloudsql_instance_postgres_log_min_messages_flag/cloudsql_instance_postgres_log_min_messages_flag.py
+3 -5
View File
@@ -17,11 +17,9 @@ class cloudsql_instance_postgres_log_min_messages_flag(Check):
findings = []
for instance in cloudsql_client.instances:
if "POSTGRES" in instance.version:
report = Check_Report_GCP(self.metadata())
report.project_id = instance.project_id
report.resource_id = instance.name
report.resource_name = instance.name
report.location = instance.region
report = Check_Report_GCP(
metadata=self.metadata(), resource_metadata=instance
)
report.status = "FAIL"
report.status_extended = f"PostgreSQL Instance {instance.name} does not have 'log_min_messages' flag set."
prowler/providers/gcp/services/cloudsql/cloudsql_instance_postgres_log_statement_flag/cloudsql_instance_postgres_log_statement_flag.py
+3 -5
View File
@@ -8,11 +8,9 @@ class cloudsql_instance_postgres_log_statement_flag(Check):
findings = []
for instance in cloudsql_client.instances:
if "POSTGRES" in instance.version:
report = Check_Report_GCP(self.metadata())
report.project_id = instance.project_id
report.resource_id = instance.name
report.resource_name = instance.name
report.location = instance.region
report = Check_Report_GCP(
metadata=self.metadata(), resource_metadata=instance
)
report.status = "FAIL"
report.status_extended = f"PostgreSQL Instance {instance.name} does not have 'log_statement' flag set to '{desired_log_statement}'."
for flag in instance.flags:
prowler/providers/gcp/services/cloudsql/cloudsql_instance_private_ip_assignment/cloudsql_instance_private_ip_assignment.py
+3 -5
View File
@@ -6,11 +6,9 @@ class cloudsql_instance_private_ip_assignment(Check):
def execute(self) -> Check_Report_GCP:
findings = []
for instance in cloudsql_client.instances:
report = Check_Report_GCP(self.metadata())
report.project_id = instance.project_id
report.resource_id = instance.name
report.resource_name = instance.name
report.location = instance.region
report = Check_Report_GCP(
metadata=self.metadata(), resource_metadata=instance
)
report.status = "PASS"
report.status_extended = f"Database Instance {instance.name} does not have private IP assignments."
for address in instance.ip_addresses:
prowler/providers/gcp/services/cloudsql/cloudsql_instance_public_access/cloudsql_instance_public_access.py
+3 -5
View File
@@ -6,11 +6,9 @@ class cloudsql_instance_public_access(Check):
def execute(self) -> Check_Report_GCP:
findings = []
for instance in cloudsql_client.instances:
report = Check_Report_GCP(self.metadata())
report.project_id = instance.project_id
report.resource_id = instance.name
report.resource_name = instance.name
report.location = instance.region
report = Check_Report_GCP(
metadata=self.metadata(), resource_metadata=instance
)
report.status = "PASS"
report.status_extended = f"Database Instance {instance.name} does not whitelist all Public IP Addresses."
for network in instance.authorized_networks:
prowler/providers/gcp/services/cloudsql/cloudsql_instance_public_ip/cloudsql_instance_public_ip.py
+3 -5
View File
@@ -6,11 +6,9 @@ class cloudsql_instance_public_ip(Check):
def execute(self) -> Check_Report_GCP:
findings = []
for instance in cloudsql_client.instances:
report = Check_Report_GCP(self.metadata())
report.project_id = instance.project_id
report.resource_id = instance.name
report.resource_name = instance.name
report.location = instance.region
report = Check_Report_GCP(
metadata=self.metadata(), resource_metadata=instance
)
report.status = "PASS"
report.status_extended = (
f"Database Instance {instance.name} does not have a public IP."
prowler/providers/gcp/services/cloudsql/cloudsql_instance_sqlserver_contained_database_authentication_flag/cloudsql_instance_sqlserver_contained_database_authentication_flag.py
+3 -5
View File
@@ -7,11 +7,9 @@ class cloudsql_instance_sqlserver_contained_database_authentication_flag(Check):
findings = []
for instance in cloudsql_client.instances:
if "SQLSERVER" in instance.version:
report = Check_Report_GCP(self.metadata())
report.project_id = instance.project_id
report.resource_id = instance.name
report.resource_name = instance.name
report.location = instance.region
report = Check_Report_GCP(
metadata=self.metadata(), resource_metadata=instance
)
report.status = "PASS"
report.status_extended = f"SQL Server Instance {instance.name} has 'contained database authentication' flag set to 'off'."
for flag in instance.flags:
prowler/providers/gcp/services/cloudsql/cloudsql_instance_sqlserver_cross_db_ownership_chaining_flag/cloudsql_instance_sqlserver_cross_db_ownership_chaining_flag.py
+3 -5
View File
@@ -7,11 +7,9 @@ class cloudsql_instance_sqlserver_cross_db_ownership_chaining_flag(Check):
findings = []
for instance in cloudsql_client.instances:
if "SQLSERVER" in instance.version:
report = Check_Report_GCP(self.metadata())
report.project_id = instance.project_id
report.resource_id = instance.name
report.resource_name = instance.name
report.location = instance.region
report = Check_Report_GCP(
metadata=self.metadata(), resource_metadata=instance
)
report.status = "PASS"
report.status_extended = f"SQL Server Instance {instance.name} has 'cross db ownership chaining' flag set to 'off'."
for flag in instance.flags:
prowler/providers/gcp/services/cloudsql/cloudsql_instance_sqlserver_external_scripts_enabled_flag/cloudsql_instance_sqlserver_external_scripts_enabled_flag.py
+3 -5
View File
@@ -7,11 +7,9 @@ class cloudsql_instance_sqlserver_external_scripts_enabled_flag(Check):
findings = []
for instance in cloudsql_client.instances:
if "SQLSERVER" in instance.version:
report = Check_Report_GCP(self.metadata())
report.project_id = instance.project_id
report.resource_id = instance.name
report.resource_name = instance.name
report.location = instance.region
report = Check_Report_GCP(
metadata=self.metadata(), resource_metadata=instance
)
report.status = "PASS"
report.status_extended = f"SQL Server Instance {instance.name} has 'external scripts enabled' flag set to 'off'."
for flag in instance.flags:
prowler/providers/gcp/services/cloudsql/cloudsql_instance_sqlserver_remote_access_flag/cloudsql_instance_sqlserver_remote_access_flag.py
+3 -5
View File
@@ -7,11 +7,9 @@ class cloudsql_instance_sqlserver_remote_access_flag(Check):
findings = []
for instance in cloudsql_client.instances:
if "SQLSERVER" in instance.version:
report = Check_Report_GCP(self.metadata())
report.project_id = instance.project_id
report.resource_id = instance.name
report.resource_name = instance.name
report.location = instance.region
report = Check_Report_GCP(
metadata=self.metadata(), resource_metadata=instance
)
report.status = "FAIL"
report.status_extended = f"SQL Server Instance {instance.name} has 'remote access' flag set to 'on'."
for flag in instance.flags:
prowler/providers/gcp/services/cloudsql/cloudsql_instance_sqlserver_trace_flag/cloudsql_instance_sqlserver_trace_flag.py
+3 -5
View File
@@ -7,11 +7,9 @@ class cloudsql_instance_sqlserver_trace_flag(Check):
findings = []
for instance in cloudsql_client.instances:
if "SQLSERVER" in instance.version:
report = Check_Report_GCP(self.metadata())
report.project_id = instance.project_id
report.resource_id = instance.name
report.resource_name = instance.name
report.location = instance.region
report = Check_Report_GCP(
metadata=self.metadata(), resource_metadata=instance
)
report.status = "FAIL"
report.status_extended = f"SQL Server Instance {instance.name} has '3625 (trace flag)' flag set to 'off'."
prowler/providers/gcp/services/cloudsql/cloudsql_instance_sqlserver_user_connections_flag/cloudsql_instance_sqlserver_user_connections_flag.py
+3 -5
View File
@@ -7,11 +7,9 @@ class cloudsql_instance_sqlserver_user_connections_flag(Check):
findings = []
for instance in cloudsql_client.instances:
if "SQLSERVER" in instance.version:
report = Check_Report_GCP(self.metadata())
report.project_id = instance.project_id
report.resource_id = instance.name
report.resource_name = instance.name
report.location = instance.region
report = Check_Report_GCP(
metadata=self.metadata(), resource_metadata=instance
)
report.status = "PASS"
report.status_extended = f"SQL Server Instance {instance.name} has 'user connections' flag set to '0'."
prowler/providers/gcp/services/cloudsql/cloudsql_instance_sqlserver_user_options_flag/cloudsql_instance_sqlserver_user_options_flag.py
+3 -5
View File
@@ -7,11 +7,9 @@ class cloudsql_instance_sqlserver_user_options_flag(Check):
findings = []
for instance in cloudsql_client.instances:
if "SQLSERVER" in instance.version:
report = Check_Report_GCP(self.metadata())
report.project_id = instance.project_id
report.resource_id = instance.name
report.resource_name = instance.name
report.location = instance.region
report = Check_Report_GCP(
metadata=self.metadata(), resource_metadata=instance
)
report.status = "PASS"
report.status_extended = f"SQL Server Instance {instance.name} does not have 'user options' flag set."
for flag in instance.flags:
prowler/providers/gcp/services/cloudsql/cloudsql_instance_ssl_connections/cloudsql_instance_ssl_connections.py
+3 -5
View File
@@ -6,11 +6,9 @@ class cloudsql_instance_ssl_connections(Check):
def execute(self) -> Check_Report_GCP:
findings = []
for instance in cloudsql_client.instances:
report = Check_Report_GCP(self.metadata())
report.project_id = instance.project_id
report.resource_id = instance.name
report.resource_name = instance.name
report.location = instance.region
report = Check_Report_GCP(
metadata=self.metadata(), resource_metadata=instance
)
report.status = "PASS"
report.status_extended = (
f"Database Instance {instance.name} requires SSL connections."
prowler/providers/gcp/services/cloudstorage/cloudstorage_bucket_log_retention_policy_lock/cloudstorage_bucket_log_retention_policy_lock.py
+3 -5
View File
@@ -15,11 +15,9 @@ class cloudstorage_bucket_log_retention_policy_lock(Check):
log_buckets.append(sink.destination.split("/")[-1])
for bucket in cloudstorage_client.buckets:
if bucket.name in log_buckets:
report = Check_Report_GCP(self.metadata())
report.project_id = bucket.project_id
report.resource_id = bucket.id
report.resource_name = bucket.name
report.location = bucket.region
report = Check_Report_GCP(
metadata=self.metadata(), resource_metadata=bucket
)
report.status = "FAIL"
report.status_extended = (
f"Log Sink Bucket {bucket.name} has no Retention Policy."
prowler/providers/gcp/services/cloudstorage/cloudstorage_bucket_public_access/cloudstorage_bucket_public_access.py
+3 -5
View File
@@ -8,11 +8,9 @@ class cloudstorage_bucket_public_access(Check):
def execute(self) -> Check_Report_GCP:
findings = []
for bucket in cloudstorage_client.buckets:
report = Check_Report_GCP(self.metadata())
report.project_id = bucket.project_id
report.resource_id = bucket.id
report.resource_name = bucket.name
report.location = bucket.region
report = Check_Report_GCP(
metadata=self.metadata(), resource_metadata=bucket
)
report.status = "PASS"
report.status_extended = f"Bucket {bucket.name} is not publicly accessible."
if bucket.public:
prowler/providers/gcp/services/cloudstorage/cloudstorage_bucket_uniform_bucket_level_access/cloudstorage_bucket_uniform_bucket_level_access.py
+3 -5
View File
@@ -8,11 +8,9 @@ class cloudstorage_bucket_uniform_bucket_level_access(Check):
def execute(self) -> Check_Report_GCP:
findings = []
for bucket in cloudstorage_client.buckets:
report = Check_Report_GCP(self.metadata())
report.project_id = bucket.project_id
report.resource_id = bucket.id
report.resource_name = bucket.name
report.location = bucket.region
report = Check_Report_GCP(
metadata=self.metadata(), resource_metadata=bucket
)
report.status = "PASS"
report.status_extended = (
f"Bucket {bucket.name} has uniform Bucket Level Access enabled."
prowler/providers/gcp/services/compute/compute_firewall_rdp_access_from_the_internet_allowed/compute_firewall_rdp_access_from_the_internet_allowed.py
+5 -5
View File
@@ -6,11 +6,11 @@ class compute_firewall_rdp_access_from_the_internet_allowed(Check):
def execute(self) -> Check_Report_GCP:
findings = []
for firewall in compute_client.firewalls:
report = Check_Report_GCP(self.metadata())
report.project_id = firewall.project_id
report.resource_id = firewall.id
report.resource_name = firewall.name
report.location = compute_client.region
report = Check_Report_GCP(
metadata=self.metadata(),
resource_metadata=firewall,
location=compute_client.region,
)
report.status = "PASS"
report.status_extended = f"Firewall {firewall.name} does not expose port 3389 (RDP) to the internet."
opened_port = False
prowler/providers/gcp/services/compute/compute_firewall_ssh_access_from_the_internet_allowed/compute_firewall_ssh_access_from_the_internet_allowed.py
+5 -5
View File
@@ -6,11 +6,11 @@ class compute_firewall_ssh_access_from_the_internet_allowed(Check):
def execute(self) -> Check_Report_GCP:
findings = []
for firewall in compute_client.firewalls:
report = Check_Report_GCP(self.metadata())
report.project_id = firewall.project_id
report.resource_id = firewall.id
report.resource_name = firewall.name
report.location = compute_client.region
report = Check_Report_GCP(
metadata=self.metadata(),
resource_metadata=firewall,
location=compute_client.region,
)
report.status = "PASS"
report.status_extended = f"Firewall {firewall.name} does not expose port 22 (SSH) to the internet."
opened_port = False
prowler/providers/gcp/services/compute/compute_instance_block_project_wide_ssh_keys_disabled/compute_instance_block_project_wide_ssh_keys_disabled.py
+3 -5
View File
@@ -6,11 +6,9 @@ class compute_instance_block_project_wide_ssh_keys_disabled(Check):
def execute(self) -> Check_Report_GCP:
findings = []
for instance in compute_client.instances:
report = Check_Report_GCP(self.metadata())
report.project_id = instance.project_id
report.resource_id = instance.id
report.resource_name = instance.name
report.location = instance.region
report = Check_Report_GCP(
metadata=self.metadata(), resource_metadata=instance
)
report.status = "FAIL"
report.status_extended = f"The VM Instance {instance.name} is making use of common/shared project-wide SSH key(s)."
if instance.metadata.get("items"):
prowler/providers/gcp/services/compute/compute_instance_confidential_computing_enabled/compute_instance_confidential_computing_enabled.py
+3 -5
View File
@@ -6,11 +6,9 @@ class compute_instance_confidential_computing_enabled(Check):
def execute(self) -> Check_Report_GCP:
findings = []
for instance in compute_client.instances:
report = Check_Report_GCP(self.metadata())
report.project_id = instance.project_id
report.resource_id = instance.id
report.resource_name = instance.name
report.location = instance.region
report = Check_Report_GCP(
metadata=self.metadata(), resource_metadata=instance
)
report.status = "PASS"
report.status_extended = (
f"VM Instance {instance.name} has Confidential Computing enabled."
prowler/providers/gcp/services/compute/compute_instance_default_service_account_in_use/compute_instance_default_service_account_in_use.py
+3 -5
View File
@@ -6,11 +6,9 @@ class compute_instance_default_service_account_in_use(Check):
def execute(self) -> Check_Report_GCP:
findings = []
for instance in compute_client.instances:
report = Check_Report_GCP(self.metadata())
report.project_id = instance.project_id
report.resource_id = instance.id
report.resource_name = instance.name
report.location = instance.region
report = Check_Report_GCP(
metadata=self.metadata(), resource_metadata=instance
)
report.status = "PASS"
report.status_extended = f"The default service account is not configured to be used with VM Instance {instance.name}."
if (
prowler/providers/gcp/services/compute/compute_instance_default_service_account_in_use_with_full_api_access/compute_instance_default_service_account_in_use_with_full_api_access.py
+3 -5
View File
@@ -6,11 +6,9 @@ class compute_instance_default_service_account_in_use_with_full_api_access(Check
def execute(self) -> Check_Report_GCP:
findings = []
for instance in compute_client.instances:
report = Check_Report_GCP(self.metadata())
report.project_id = instance.project_id
report.resource_id = instance.id
report.resource_name = instance.name
report.location = instance.region
report = Check_Report_GCP(
metadata=self.metadata(), resource_metadata=instance
)
report.status = "PASS"
report.status_extended = f"The VM Instance {instance.name} is not configured to use the default service account with full access to all cloud APIs."
for service_account in instance.service_accounts:
prowler/providers/gcp/services/compute/compute_instance_encryption_with_csek_enabled/compute_instance_encryption_with_csek_enabled.py
+3 -5
View File
@@ -6,11 +6,9 @@ class compute_instance_encryption_with_csek_enabled(Check):
def execute(self) -> Check_Report_GCP:
findings = []
for instance in compute_client.instances:
report = Check_Report_GCP(self.metadata())
report.project_id = instance.project_id
report.resource_id = instance.id
report.resource_name = instance.name
report.location = instance.region
report = Check_Report_GCP(
metadata=self.metadata(), resource_metadata=instance
)
report.status = "FAIL"
report.status_extended = f"The VM Instance {instance.name} has the following unencrypted disks: '{', '.join([i[0] for i in instance.disks_encryption if not i[1]])}'."
if all([i[1] for i in instance.disks_encryption]):
prowler/providers/gcp/services/compute/compute_instance_ip_forwarding_is_enabled/compute_instance_ip_forwarding_is_enabled.py
+3 -5
View File
@@ -6,11 +6,9 @@ class compute_instance_ip_forwarding_is_enabled(Check):
def execute(self) -> Check_Report_GCP:
findings = []
for instance in compute_client.instances:
report = Check_Report_GCP(self.metadata())
report.project_id = instance.project_id
report.resource_id = instance.id
report.resource_name = instance.name
report.location = instance.region
report = Check_Report_GCP(
metadata=self.metadata(), resource_metadata=instance
)
report.status = "PASS"
report.status_extended = (
f"The IP Forwarding of VM Instance {instance.name} is not enabled."
prowler/providers/gcp/services/compute/compute_instance_public_ip/compute_instance_public_ip.py
+3 -5
View File
@@ -6,11 +6,9 @@ class compute_instance_public_ip(Check):
def execute(self) -> Check_Report_GCP:
findings = []
for instance in compute_client.instances:
report = Check_Report_GCP(self.metadata())
report.project_id = instance.project_id
report.resource_id = instance.id
report.resource_name = instance.name
report.location = instance.region
report = Check_Report_GCP(
metadata=self.metadata(), resource_metadata=instance
)
report.status = "PASS"
report.status_extended = (
f"VM Instance {instance.name} does not have a public IP."
prowler/providers/gcp/services/compute/compute_instance_serial_ports_in_use/compute_instance_serial_ports_in_use.py
+3 -5
View File
@@ -6,11 +6,9 @@ class compute_instance_serial_ports_in_use(Check):
def execute(self) -> Check_Report_GCP:
findings = []
for instance in compute_client.instances:
report = Check_Report_GCP(self.metadata())
report.project_id = instance.project_id
report.resource_id = instance.id
report.resource_name = instance.name
report.location = instance.region
report = Check_Report_GCP(
metadata=self.metadata(), resource_metadata=instance
)
report.status = "PASS"
report.status_extended = f"VM Instance {instance.name} has Enable Connecting to Serial Ports off."
if instance.metadata.get("items"):
prowler/providers/gcp/services/compute/compute_instance_shielded_vm_enabled/compute_instance_shielded_vm_enabled.py
+3 -5
View File
@@ -6,11 +6,9 @@ class compute_instance_shielded_vm_enabled(Check):
def execute(self) -> Check_Report_GCP:
findings = []
for instance in compute_client.instances:
report = Check_Report_GCP(self.metadata())
report.project_id = instance.project_id
report.resource_id = instance.id
report.resource_name = instance.name
report.location = instance.region
report = Check_Report_GCP(
metadata=self.metadata(), resource_metadata=instance
)
report.status = "PASS"
report.status_extended = f"VM Instance {instance.name} has vTPM or Integrity Monitoring set to on."
if (
prowler/providers/gcp/services/compute/compute_loadbalancer_logging_enabled/compute_loadbalancer_logging_enabled.py
+5 -5
View File
@@ -6,11 +6,11 @@ class compute_loadbalancer_logging_enabled(Check):
def execute(self) -> Check_Report_GCP:
findings = []
for lb in compute_client.load_balancers:
report = Check_Report_GCP(self.metadata())
report.project_id = lb.project_id
report.resource_id = lb.id
report.resource_name = lb.name
report.location = compute_client.region
report = Check_Report_GCP(
metadata=self.metadata(),
resource_metadata=lb,
location=compute_client.region,
)
report.status = "PASS"
report.status_extended = f"LoadBalancer {lb.name} has logging enabled."
if not lb.logging:
prowler/providers/gcp/services/compute/compute_network_default_in_use/compute_network_default_in_use.py
+11 -8
View File
@@ -5,26 +5,29 @@ from prowler.providers.gcp.services.compute.compute_client import compute_client
class compute_network_default_in_use(Check):
def execute(self) -> Check_Report_GCP:
findings = []
projects_with_default_network = set()
projects_with_default_network = {}
# Identify projects with the default network
for network in compute_client.networks:
if network.name == "default":
projects_with_default_network.add(network.project_id)
projects_with_default_network[network.project_id] = network
# Generate reports for all projects
for project in compute_client.project_ids:
report = Check_Report_GCP(self.metadata())
report.project_id = project
report.resource_id = "default"
report.resource_name = "default"
report.location = compute_client.region
report = Check_Report_GCP(
metadata=self.metadata(),
resource_metadata=compute_client.projects[project],
project_id=project,
resource_id="default",
resource_name="default",
location=compute_client.region,
)
if project in projects_with_default_network:
report.status = "FAIL"
report.status_extended = (
f"Default network is in use in project {project}."
)
report.resource_metadata = projects_with_default_network[project]
else:
report.status = "PASS"
report.status_extended = (
prowler/providers/gcp/services/compute/compute_network_dns_logging_enabled/compute_network_dns_logging_enabled.py
+5 -5
View File
@@ -7,11 +7,11 @@ class compute_network_dns_logging_enabled(Check):
def execute(self) -> Check_Report_GCP:
findings = []
for network in compute_client.networks:
report = Check_Report_GCP(self.metadata())
report.project_id = network.project_id
report.resource_id = network.id
report.resource_name = network.name
report.location = compute_client.region
report = Check_Report_GCP(
metadata=self.metadata(),
resource_metadata=network,
location=compute_client.region,
)
report.status = "FAIL"
report.status_extended = (
f"Network {network.name} does not have DNS logging enabled."
prowler/providers/gcp/services/compute/compute_network_not_legacy/compute_network_not_legacy.py
+5 -5
View File
@@ -6,11 +6,11 @@ class compute_network_not_legacy(Check):
def execute(self) -> Check_Report_GCP:
findings = []
for network in compute_client.networks:
report = Check_Report_GCP(self.metadata())
report.project_id = network.project_id
report.resource_id = network.id
report.resource_name = network.name
report.location = compute_client.region
report = Check_Report_GCP(
metadata=self.metadata(),
resource_metadata=network,
location=compute_client.region,
)
report.status = "PASS"
report.status_extended = f"Network {network.name} is not legacy."
if network.subnet_mode == "legacy":
prowler/providers/gcp/services/compute/compute_project_os_login_enabled/compute_project_os_login_enabled.py
+6 -4
View File
@@ -6,10 +6,12 @@ class compute_project_os_login_enabled(Check):
def execute(self) -> Check_Report_GCP:
findings = []
for project in compute_client.projects:
report = Check_Report_GCP(self.metadata())
report.project_id = project.id
report.resource_id = project.id
report.location = compute_client.region
report = Check_Report_GCP(
metadata=self.metadata(),
resource_metadata=project,
project_id=project.id,
location=compute_client.region,
)
report.status = "PASS"
report.status_extended = f"Project {project.id} has OS Login enabled."
if not project.enable_oslogin:
prowler/providers/gcp/services/compute/compute_public_address_shodan/compute_public_address_shodan.py
+3 -4
View File
@@ -13,10 +13,9 @@ class compute_public_address_shodan(Check):
api = shodan.Shodan(shodan_api_key)
for address in compute_client.addresses:
if address.type == "EXTERNAL":
report = Check_Report_GCP(self.metadata())
report.project_id = address.project_id
report.resource_id = address.id
report.location = address.region
report = Check_Report_GCP(
metadata=self.metadata(), resource_metadata=address
)
try:
shodan_info = api.host(address.ip)
report.status = "FAIL"
prowler/providers/gcp/services/compute/compute_service.py
+2 -2
View File
@@ -15,7 +15,7 @@ class Compute(GCPService):
self.subnets = []
self.addresses = []
self.firewalls = []
self.projects = []
self.compute_projects = []
self.load_balancers = []
self._get_url_maps()
self._describe_backend_service()
@@ -72,7 +72,7 @@ class Compute(GCPService):
for item in response["commonInstanceMetadata"].get("items", []):
if item["key"] == "enable-oslogin" and item["value"] == "TRUE":
enable_oslogin = True
self.projects.append(
self.compute_projects.append(
Project(id=project_id, enable_oslogin=enable_oslogin)
)
except Exception as error:
prowler/providers/gcp/services/compute/compute_subnet_flow_logs_enabled/compute_subnet_flow_logs_enabled.py
+3 -5
View File
@@ -6,11 +6,9 @@ class compute_subnet_flow_logs_enabled(Check):
def execute(self) -> Check_Report_GCP:
findings = []
for subnet in compute_client.subnets:
report = Check_Report_GCP(self.metadata())
report.project_id = subnet.project_id
report.resource_id = subnet.id
report.resource_name = subnet.name
report.location = subnet.region
report = Check_Report_GCP(
metadata=self.metadata(), resource_metadata=subnet
)
report.status = "PASS"
report.status_extended = f"Subnet {subnet.name} in network {subnet.network} has flow logs enabled."
if not subnet.flow_logs:
prowler/providers/gcp/services/dataproc/dataproc_encrypted_with_cmks_disabled/dataproc_encrypted_with_cmks_disabled.py
+5 -5
View File
@@ -6,11 +6,11 @@ class dataproc_encrypted_with_cmks_disabled(Check):
def execute(self) -> Check_Report_GCP:
findings = []
for cluster in dataproc_client.clusters:
report = Check_Report_GCP(self.metadata())
report.project_id = cluster.project_id
report.resource_id = cluster.id
report.resource_name = cluster.name
report.location = dataproc_client.region
report = Check_Report_GCP(
metadata=self.metadata(),
resource_metadata=cluster,
location=dataproc_client.region,
)
report.status = "PASS"
report.status_extended = f"Dataproc cluster {cluster.name} is encrypted with customer managed encryption keys."
if cluster.encryption_config.get("gcePdKmsKeyName") is None:
prowler/providers/gcp/services/dns/dns_dnssec_disabled/dns_dnssec_disabled.py
+5 -5
View File
@@ -6,11 +6,11 @@ class dns_dnssec_disabled(Check):
def execute(self) -> Check_Report_GCP:
findings = []
for managed_zone in dns_client.managed_zones:
report = Check_Report_GCP(self.metadata())
report.project_id = managed_zone.project_id
report.resource_id = managed_zone.id
report.resource_name = managed_zone.name
report.location = dns_client.region
report = Check_Report_GCP(
metadata=self.metadata(),
resource_metadata=managed_zone,
location=dns_client.region,
)
report.status = "PASS"
report.status_extended = (
f"Cloud DNS {managed_zone.name} has DNSSEC enabled."
prowler/providers/gcp/services/dns/dns_rsasha1_in_use_to_key_sign_in_dnssec/dns_rsasha1_in_use_to_key_sign_in_dnssec.py
+5 -5
View File
@@ -6,11 +6,11 @@ class dns_rsasha1_in_use_to_key_sign_in_dnssec(Check):
def execute(self) -> Check_Report_GCP:
findings = []
for managed_zone in dns_client.managed_zones:
report = Check_Report_GCP(self.metadata())
report.project_id = managed_zone.project_id
report.resource_id = managed_zone.id
report.resource_name = managed_zone.name
report.location = dns_client.region
report = Check_Report_GCP(
metadata=self.metadata(),
resource_metadata=managed_zone,
location=dns_client.region,
)
report.status = "PASS"
report.status_extended = f"Cloud DNS {managed_zone.name} is not using RSASHA1 algorithm as key signing."
if any(
prowler/providers/gcp/services/dns/dns_rsasha1_in_use_to_zone_sign_in_dnssec/dns_rsasha1_in_use_to_zone_sign_in_dnssec.py
+5 -5
View File
@@ -6,11 +6,11 @@ class dns_rsasha1_in_use_to_zone_sign_in_dnssec(Check):
def execute(self) -> Check_Report_GCP:
findings = []
for managed_zone in dns_client.managed_zones:
report = Check_Report_GCP(self.metadata())
report.project_id = managed_zone.project_id
report.resource_id = managed_zone.id
report.resource_name = managed_zone.name
report.location = dns_client.region
report = Check_Report_GCP(
metadata=self.metadata(),
resource_metadata=managed_zone,
location=dns_client.region,
)
report.status = "PASS"
report.status_extended = f"Cloud DNS {managed_zone.name} is not using RSASHA1 algorithm as zone signing."
if any(
prowler/providers/gcp/services/gcr/gcr_container_scanning_enabled/gcr_container_scanning_enabled.py
+8 -5
View File
@@ -8,11 +8,14 @@ class gcr_container_scanning_enabled(Check):
def execute(self) -> Check_Report_GCP:
findings = []
for project_id in serviceusage_client.project_ids:
report = Check_Report_GCP(self.metadata())
report.project_id = project_id
report.resource_id = "containerscanning.googleapis.com"
report.resource_name = "GCR Container Scanning"
report.location = serviceusage_client.region
report = Check_Report_GCP(
metadata=self.metadata(),
resource_metadata=serviceusage_client.projects[project_id],
resource_id="containerscanning.googleapis.com",
resource_name="GCR Container Scanning",
project_id=project_id,
location=serviceusage_client.region,
)
report.status = "FAIL"
report.status_extended = (
f"GCR Container Scanning is not enabled in project {project_id}."
prowler/providers/gcp/services/gke/gke_cluster_no_default_service_account/gke_cluster_no_default_service_account.py
+3 -5
View File
@@ -6,11 +6,9 @@ class gke_cluster_no_default_service_account(Check):
def execute(self) -> Check_Report_GCP:
findings = []
for cluster in gke_client.clusters.values():
report = Check_Report_GCP(self.metadata())
report.project_id = cluster.project_id
report.resource_id = cluster.id
report.resource_name = cluster.name
report.location = cluster.region
report = Check_Report_GCP(
metadata=self.metadata(), resource_metadata=cluster
)
report.status = "PASS"
report.status_extended = f"GKE cluster {cluster.name} is not using the Compute Engine default service account."
if not cluster.node_pools and cluster.service_account == "default":
prowler/providers/gcp/services/iam/iam_account_access_approval_enabled/iam_account_access_approval_enabled.py
+6 -4
View File
@@ -8,10 +8,12 @@ class iam_account_access_approval_enabled(Check):
def execute(self) -> Check_Report_GCP:
findings = []
for project_id in accessapproval_client.project_ids:
report = Check_Report_GCP(self.metadata())
report.project_id = project_id
report.resource_id = project_id
report.location = accessapproval_client.region
report = Check_Report_GCP(
metadata=self.metadata(),
resource_metadata=accessapproval_client.projects[project_id],
project_id=project_id,
location=accessapproval_client.region,
)
report.status = "PASS"
report.status_extended = (
f"Project {project_id} has Access Approval enabled."
prowler/providers/gcp/services/iam/iam_audit_logs_enabled/iam_audit_logs_enabled.py
+6 -4
View File
@@ -8,10 +8,12 @@ class iam_audit_logs_enabled(Check):
def execute(self) -> Check_Report_GCP:
findings = []
for project in cloudresourcemanager_client.projects:
report = Check_Report_GCP(self.metadata())
report.project_id = project.id
report.location = cloudresourcemanager_client.region
report.resource_id = project.id
report = Check_Report_GCP(
metadata=self.metadata(),
resource_metadata=project,
project_id=project.id,
location=cloudresourcemanager_client.region,
)
report.status = "PASS"
report.status_extended = f"Audit Logs are enabled for project {project.id}."
if not project.audit_logging:
prowler/providers/gcp/services/iam/iam_cloud_asset_inventory_enabled/iam_cloud_asset_inventory_enabled.py
+8 -5
View File
@@ -8,11 +8,14 @@ class iam_cloud_asset_inventory_enabled(Check):
def execute(self) -> Check_Report_GCP:
findings = []
for project_id in serviceusage_client.project_ids:
report = Check_Report_GCP(self.metadata())
report.project_id = project_id
report.resource_id = "cloudasset.googleapis.com"
report.resource_name = "Cloud Asset Inventory"
report.location = serviceusage_client.region
report = Check_Report_GCP(
metadata=self.metadata(),
resource_metadata=serviceusage_client.projects[project_id],
resource_id="cloudasset.googleapis.com",
resource_name="Cloud Asset Inventory",
project_id=project_id,
location=serviceusage_client.region,
)
report.status = "FAIL"
report.status_extended = (
f"Cloud Asset Inventory is not enabled in project {project_id}."
prowler/providers/gcp/services/iam/iam_no_service_roles_at_project_level/iam_no_service_roles_at_project_level.py
+14 -10
View File
@@ -9,11 +9,13 @@ class iam_no_service_roles_at_project_level(Check):
findings = []
failed_projects = set()
for binding in cloudresourcemanager_client.bindings:
report = Check_Report_GCP(self.metadata())
report.project_id = binding.project_id
report.resource_id = binding.role
report.resource_name = binding.role
report.location = cloudresourcemanager_client.region
report = Check_Report_GCP(
metadata=self.metadata(),
resource_metadata=binding,
resource_id=binding.role,
resource_name=binding.role,
location=cloudresourcemanager_client.region,
)
if binding.role in [
"roles/iam.serviceAccountUser",
"roles/iam.serviceAccountTokenCreator",
@@ -25,12 +27,14 @@ class iam_no_service_roles_at_project_level(Check):
for project in cloudresourcemanager_client.project_ids:
if project not in failed_projects:
report = Check_Report_GCP(self.metadata())
report.project_id = project
report.resource_id = project
report.resource_name = project
report = Check_Report_GCP(
metadata=self.metadata(),
resource_metadata=cloudresourcemanager_client.projects[project],
project_id=project,
resource_name=project,
location=cloudresourcemanager_client.region,
)
report.status = "PASS"
report.location = cloudresourcemanager_client.region
report.status_extended = f"No IAM Users assigned to service roles at project level {project}."
findings.append(report)
return findings
prowler/providers/gcp/services/iam/iam_organization_essential_contacts_configured/iam_organization_essential_contacts_configured.py
+6 -5
View File
@@ -8,11 +8,12 @@ class iam_organization_essential_contacts_configured(Check):
def execute(self) -> Check_Report_GCP:
findings = []
for org in essentialcontacts_client.organizations:
report = Check_Report_GCP(self.metadata())
report.project_id = essentialcontacts_client.default_project_id
report.resource_id = org.id
report.resource_name = org.name
report.location = essentialcontacts_client.region
report = Check_Report_GCP(
metadata=self.metadata(),
resource_metadata=org,
project_id=essentialcontacts_client.default_project_id,
location=essentialcontacts_client.region,
)
report.status = "FAIL"
report.status_extended = (
f"Organization {org.name} does not have essential contacts configured."
prowler/providers/gcp/services/iam/iam_role_kms_enforce_separation_of_duties/iam_role_kms_enforce_separation_of_duties.py
+6 -4
View File
@@ -10,10 +10,12 @@ class iam_role_kms_enforce_separation_of_duties(Check):
for project in cloudresourcemanager_client.project_ids:
non_compliant_members = []
kms_admin_members = []
report = Check_Report_GCP(self.metadata())
report.project_id = project
report.location = cloudresourcemanager_client.region
report.resource_id = project
report = Check_Report_GCP(
metadata=self.metadata(),
resource_metadata=cloudresourcemanager_client.projects[project],
project_id=project,
location=cloudresourcemanager_client.region,
)
report.status = "PASS"
report.status_extended = f"Principle of separation of duties was enforced for KMS-Related Roles in project {project}."
for binding in cloudresourcemanager_client.bindings:
prowler/providers/gcp/services/iam/iam_role_sa_enforce_separation_of_duties/iam_role_sa_enforce_separation_of_duties.py
+6 -4
View File
@@ -9,10 +9,12 @@ class iam_role_sa_enforce_separation_of_duties(Check):
findings = []
for project in cloudresourcemanager_client.project_ids:
non_compliant_members = []
report = Check_Report_GCP(self.metadata())
report.project_id = project
report.location = cloudresourcemanager_client.region
report.resource_id = project
report = Check_Report_GCP(
metadata=self.metadata(),
resource_metadata=cloudresourcemanager_client.projects[project],
location=cloudresourcemanager_client.region,
project_id=project,
)
report.status = "PASS"
report.status_extended = f"Principle of separation of duties was enforced for Service-Account Related Roles in project {project}."
for binding in cloudresourcemanager_client.bindings:
prowler/providers/gcp/services/iam/iam_sa_no_administrative_privileges/iam_sa_no_administrative_privileges.py
+6 -5
View File
@@ -9,11 +9,12 @@ class iam_sa_no_administrative_privileges(Check):
def execute(self) -> Check_Report_GCP:
findings = []
for account in iam_client.service_accounts:
report = Check_Report_GCP(self.metadata())
report.project_id = account.project_id
report.resource_id = account.email
report.resource_name = account.name
report.location = iam_client.region
report = Check_Report_GCP(
metadata=self.metadata(),
resource_metadata=account,
resource_id=account.email,
location=iam_client.region,
)
report.status = "PASS"
report.status_extended = (
f"Account {account.email} has no administrative privileges."
prowler/providers/gcp/services/iam/iam_sa_no_user_managed_keys/iam_sa_no_user_managed_keys.py
+6 -5
View File
@@ -6,11 +6,12 @@ class iam_sa_no_user_managed_keys(Check):
def execute(self) -> Check_Report_GCP:
findings = []
for account in iam_client.service_accounts:
report = Check_Report_GCP(self.metadata())
report.project_id = account.project_id
report.resource_id = account.email
report.resource_name = account.name
report.location = iam_client.region
report = Check_Report_GCP(
metadata=self.metadata(),
resource_metadata=account,
resource_id=account.email,
location=iam_client.region,
)
report.status = "PASS"
report.status_extended = (
f"Account {account.email} does not have user-managed keys."
prowler/providers/gcp/services/iam/iam_sa_user_managed_key_rotate_90_days/iam_sa_user_managed_key_rotate_90_days.py
+7 -5
View File
@@ -11,11 +11,13 @@ class iam_sa_user_managed_key_rotate_90_days(Check):
for key in account.keys:
if key.type == "USER_MANAGED":
last_rotated = (datetime.now() - key.valid_after).days
report = Check_Report_GCP(self.metadata())
report.project_id = account.project_id
report.resource_id = key.name
report.resource_name = account.email
report.location = iam_client.region
report = Check_Report_GCP(
metadata=self.metadata(),
resource_metadata=account,
resource_id=key.name,
resource_name=account.email,
location=iam_client.region,
)
report.status = "PASS"
report.status_extended = f"User-managed key {key.name} for account {account.email} was rotated over the last 90 days ({last_rotated} days ago)."
if last_rotated > 90:
prowler/providers/gcp/services/kms/kms_key_not_publicly_accessible/kms_key_not_publicly_accessible.py
+1 -5
View File
@@ -6,11 +6,7 @@ class kms_key_not_publicly_accessible(Check):
def execute(self) -> Check_Report_GCP:
findings = []
for key in kms_client.crypto_keys:
report = Check_Report_GCP(self.metadata())
report.project_id = key.project_id
report.resource_id = key.id
report.resource_name = key.name
report.location = key.location
report = Check_Report_GCP(metadata=self.metadata(), resource_metadata=key)
report.status = "PASS"
report.status_extended = f"Key {key.name} is not exposed to Public."
for member in key.members:
prowler/providers/gcp/services/kms/kms_key_rotation_enabled/kms_key_rotation_enabled.py
+1 -5
View File
@@ -8,11 +8,7 @@ class kms_key_rotation_enabled(Check):
def execute(self) -> Check_Report_GCP:
findings = []
for key in kms_client.crypto_keys:
report = Check_Report_GCP(self.metadata())
report.project_id = key.project_id
report.resource_id = key.id
report.resource_name = key.name
report.location = key.location
report = Check_Report_GCP(metadata=self.metadata(), resource_metadata=key)
now = datetime.datetime.now()
condition_next_rotation_time = False
if key.next_rotation_time:
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.py
+11 -10
View File
@@ -16,12 +16,12 @@ class logging_log_metric_filter_and_alert_for_audit_configuration_changes_enable
'protoPayload.methodName="SetIamPolicy" AND protoPayload.serviceData.policyDelta.auditConfigDeltas:*'
in metric.filter
):
report = Check_Report_GCP(self.metadata())
report = Check_Report_GCP(
metadata=self.metadata(),
resource_metadata=metric,
location=logging_client.region,
)
projects_with_metric.add(metric.project_id)
report.project_id = metric.project_id
report.resource_id = metric.name
report.resource_name = metric.name
report.location = logging_client.region
report.status = "FAIL"
report.status_extended = f"Log metric filter {metric.name} found but no alerts associated in project {metric.project_id}."
for alert_policy in monitoring_client.alert_policies:
@@ -34,11 +34,12 @@ class logging_log_metric_filter_and_alert_for_audit_configuration_changes_enable
for project in logging_client.project_ids:
if project not in projects_with_metric:
report = Check_Report_GCP(self.metadata())
report.project_id = project
report.resource_id = project
report.resource_name = ""
report.location = logging_client.region
report = Check_Report_GCP(
metadata=self.metadata(),
resource_metadata=logging_client.projects[project],
project_id=project,
location=logging_client.region,
)
report.status = "FAIL"
report.status_extended = f"There are no log metric filters or alerts associated in project {project}."
findings.append(report)
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.py
+11 -10
View File
@@ -14,12 +14,12 @@ class logging_log_metric_filter_and_alert_for_bucket_permission_changes_enabled(
'resource.type="gcs_bucket" AND protoPayload.methodName="storage.setIamPermissions"'
in metric.filter
):
report = Check_Report_GCP(self.metadata())
report = Check_Report_GCP(
metadata=self.metadata(),
resource_metadata=metric,
location=logging_client.region,
)
projects_with_metric.add(metric.project_id)
report.project_id = metric.project_id
report.resource_id = metric.name
report.resource_name = metric.name
report.location = logging_client.region
report.status = "FAIL"
report.status_extended = f"Log metric filter {metric.name} found but no alerts associated in project {metric.project_id}."
for alert_policy in monitoring_client.alert_policies:
@@ -32,11 +32,12 @@ class logging_log_metric_filter_and_alert_for_bucket_permission_changes_enabled(
for project in logging_client.project_ids:
if project not in projects_with_metric:
report = Check_Report_GCP(self.metadata())
report.project_id = project
report.resource_id = project
report.resource_name = ""
report.location = logging_client.region
report = Check_Report_GCP(
metadata=self.metadata(),
resource_metadata=logging_client.projects[project],
project_id=project,
location=logging_client.region,
)
report.status = "FAIL"
report.status_extended = f"There are no log metric filters or alerts associated in project {project}."
findings.append(report)
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.py
+11 -10
View File
@@ -14,12 +14,12 @@ class logging_log_metric_filter_and_alert_for_custom_role_changes_enabled(Check)
'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")'
in metric.filter
):
report = Check_Report_GCP(self.metadata())
report = Check_Report_GCP(
metadata=self.metadata(),
resource_metadata=metric,
location=logging_client.region,
)
projects_with_metric.add(metric.project_id)
report.project_id = metric.project_id
report.resource_id = metric.name
report.resource_name = metric.name
report.location = logging_client.region
report.status = "FAIL"
report.status_extended = f"Log metric filter {metric.name} found but no alerts associated in project {metric.project_id}."
for alert_policy in monitoring_client.alert_policies:
@@ -32,11 +32,12 @@ class logging_log_metric_filter_and_alert_for_custom_role_changes_enabled(Check)
for project in logging_client.project_ids:
if project not in projects_with_metric:
report = Check_Report_GCP(self.metadata())
report.project_id = project
report.resource_id = project
report.resource_name = ""
report.location = logging_client.region
report = Check_Report_GCP(
metadata=self.metadata(),
resource_metadata=logging_client.projects[project],
project_id=project,
location=logging_client.region,
)
report.status = "FAIL"
report.status_extended = f"There are no log metric filters or alerts associated in project {project}."
findings.append(report)
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.py
+11 -10
View File
@@ -14,12 +14,12 @@ class logging_log_metric_filter_and_alert_for_project_ownership_changes_enabled(
'(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")'
in metric.filter
):
report = Check_Report_GCP(self.metadata())
report = Check_Report_GCP(
metadata=self.metadata(),
resource_metadata=metric,
location=logging_client.region,
)
projects_with_metric.add(metric.project_id)
report.project_id = metric.project_id
report.resource_id = metric.name
report.resource_name = metric.name
report.location = logging_client.region
report.status = "FAIL"
report.status_extended = f"Log metric filter {metric.name} found but no alerts associated in project {metric.project_id}."
for alert_policy in monitoring_client.alert_policies:
@@ -32,11 +32,12 @@ class logging_log_metric_filter_and_alert_for_project_ownership_changes_enabled(
for project in logging_client.project_ids:
if project not in projects_with_metric:
report = Check_Report_GCP(self.metadata())
report.project_id = project
report.resource_id = project
report.resource_name = ""
report.location = logging_client.region
report = Check_Report_GCP(
metadata=self.metadata(),
resource_metadata=logging_client.projects[project],
project_id=project,
location=logging_client.region,
)
report.status = "FAIL"
report.status_extended = f"There are no log metric filters or alerts associated in project {project}."
findings.append(report)
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.py
+11 -10
View File
@@ -13,12 +13,12 @@ class logging_log_metric_filter_and_alert_for_sql_instance_configuration_changes
projects_with_metric = set()
for metric in logging_client.metrics:
if 'protoPayload.methodName="cloudsql.instances.update"' in metric.filter:
report = Check_Report_GCP(self.metadata())
report = Check_Report_GCP(
metadata=self.metadata(),
resource_metadata=metric,
location=logging_client.region,
)
projects_with_metric.add(metric.project_id)
report.project_id = metric.project_id
report.resource_id = metric.name
report.resource_name = metric.name
report.location = logging_client.region
report.status = "FAIL"
report.status_extended = f"Log metric filter {metric.name} found but no alerts associated in project {metric.project_id}."
for alert_policy in monitoring_client.alert_policies:
@@ -31,11 +31,12 @@ class logging_log_metric_filter_and_alert_for_sql_instance_configuration_changes
for project in logging_client.project_ids:
if project not in projects_with_metric:
report = Check_Report_GCP(self.metadata())
report.project_id = project
report.resource_id = project
report.resource_name = ""
report.location = logging_client.region
report = Check_Report_GCP(
metadata=self.metadata(),
resource_metadata=logging_client.projects[project],
project_id=project,
location=logging_client.region,
)
report.status = "FAIL"
report.status_extended = f"There are no log metric filters or alerts associated in project {project}."
findings.append(report)
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.py
+11 -10
View File
@@ -14,12 +14,12 @@ class logging_log_metric_filter_and_alert_for_vpc_firewall_rule_changes_enabled(
'resource.type="gce_firewall_rule" AND (protoPayload.methodName:"compute.firewalls.patch" OR protoPayload.methodName:"compute.firewalls.insert" OR protoPayload.methodName:"compute.firewalls.delete")'
in metric.filter
):
report = Check_Report_GCP(self.metadata())
report = Check_Report_GCP(
metadata=self.metadata(),
resource_metadata=metric,
location=logging_client.region,
)
projects_with_metric.add(metric.project_id)
report.project_id = metric.project_id
report.resource_id = metric.name
report.resource_name = metric.name
report.location = logging_client.region
report.status = "FAIL"
report.status_extended = f"Log metric filter {metric.name} found but no alerts associated in project {metric.project_id}."
for alert_policy in monitoring_client.alert_policies:
@@ -32,11 +32,12 @@ class logging_log_metric_filter_and_alert_for_vpc_firewall_rule_changes_enabled(
for project in logging_client.project_ids:
if project not in projects_with_metric:
report = Check_Report_GCP(self.metadata())
report.project_id = project
report.resource_id = project
report.resource_name = ""
report.location = logging_client.region
report = Check_Report_GCP(
metadata=self.metadata(),
resource_metadata=logging_client.projects[project],
project_id=project,
location=logging_client.region,
)
report.status = "FAIL"
report.status_extended = f"There are no log metric filters or alerts associated in project {project}."
findings.append(report)
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.py
+11 -10
View File
@@ -14,12 +14,12 @@ class logging_log_metric_filter_and_alert_for_vpc_network_changes_enabled(Check)
'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")'
in metric.filter
):
report = Check_Report_GCP(self.metadata())
report = Check_Report_GCP(
metadata=self.metadata(),
resource_metadata=metric,
location=logging_client.region,
)
projects_with_metric.add(metric.project_id)
report.project_id = metric.project_id
report.resource_id = metric.name
report.resource_name = metric.name
report.location = logging_client.region
report.status = "FAIL"
report.status_extended = f"Log metric filter {metric.name} found but no alerts associated in project {metric.project_id}."
for alert_policy in monitoring_client.alert_policies:
@@ -32,11 +32,12 @@ class logging_log_metric_filter_and_alert_for_vpc_network_changes_enabled(Check)
for project in logging_client.project_ids:
if project not in projects_with_metric:
report = Check_Report_GCP(self.metadata())
report.project_id = project
report.resource_id = project
report.resource_name = ""
report.location = logging_client.region
report = Check_Report_GCP(
metadata=self.metadata(),
resource_metadata=logging_client.projects[project],
project_id=project,
location=logging_client.region,
)
report.status = "FAIL"
report.status_extended = f"There are no log metric filters or alerts associated in project {project}."
findings.append(report)
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.py
+11 -10
View File
@@ -14,12 +14,12 @@ class logging_log_metric_filter_and_alert_for_vpc_network_route_changes_enabled(
'resource.type="gce_route" AND (protoPayload.methodName:"compute.routes.delete" OR protoPayload.methodName:"compute.routes.insert")'
in metric.filter
):
report = Check_Report_GCP(self.metadata())
report = Check_Report_GCP(
metadata=self.metadata(),
resource_metadata=metric,
location=logging_client.region,
)
projects_with_metric.add(metric.project_id)
report.project_id = metric.project_id
report.resource_id = metric.name
report.resource_name = metric.name
report.location = logging_client.region
report.status = "FAIL"
report.status_extended = f"Log metric filter {metric.name} found but no alerts associated in project {metric.project_id}."
for alert_policy in monitoring_client.alert_policies:
@@ -32,11 +32,12 @@ class logging_log_metric_filter_and_alert_for_vpc_network_route_changes_enabled(
for project in logging_client.project_ids:
if project not in projects_with_metric:
report = Check_Report_GCP(self.metadata())
report.project_id = project
report.resource_id = project
report.resource_name = ""
report.location = logging_client.region
report = Check_Report_GCP(
metadata=self.metadata(),
resource_metadata=logging_client.projects[project],
project_id=project,
location=logging_client.region,
)
report.status = "FAIL"
report.status_extended = f"There are no log metric filters or alerts associated in project {project}."
findings.append(report)
prowler/providers/gcp/services/logging/logging_sink_created/logging_sink_created.py
+11 -10
View File
@@ -7,12 +7,12 @@ class logging_sink_created(Check):
findings = []
projects_with_sink = set()
for sink in logging_client.sinks:
report = Check_Report_GCP(self.metadata())
report = Check_Report_GCP(
metadata=self.metadata(),
resource_metadata=sink,
location=logging_client.region,
)
projects_with_sink.add(sink.project_id)
report.project_id = sink.project_id
report.resource_id = sink.name
report.resource_name = sink.name
report.location = logging_client.region
report.status = "FAIL"
report.status_extended = f"Sink {sink.name} is enabled but not exporting copies of all the log entries in project {sink.project_id}."
if sink.filter == "all":
@@ -22,11 +22,12 @@ class logging_sink_created(Check):
for project in logging_client.project_ids:
if project not in projects_with_sink:
report = Check_Report_GCP(self.metadata())
report.project_id = project
report.resource_id = project
report.resource_name = ""
report.location = logging_client.region
report = Check_Report_GCP(
metadata=self.metadata(),
resource_metadata=logging_client.projects[project],
project_id=project,
location=logging_client.region,
)
report.status = "FAIL"
report.status_extended = f"There are no logging sinks to export copies of all the log entries in project {project}."
findings.append(report)
tests/providers/gcp/gcp_fixtures.py
+10 -1
View File
@@ -3,7 +3,7 @@ from uuid import uuid4
from mock import MagicMock
from prowler.providers.gcp.gcp_provider import GcpProvider
from prowler.providers.gcp.models import GCPIdentityInfo
from prowler.providers.gcp.models import GCPIdentityInfo, GCPProject
GCP_PROJECT_ID = "123456789012"
@@ -18,6 +18,15 @@ def set_mocked_gcp_provider(
provider.type = "gcp"
provider.session = MagicMock()
provider.session._service_account_email = "test@test.com"
provider.projects = {
GCP_PROJECT_ID: GCPProject(
id=GCP_PROJECT_ID,
number="123456789012",
name="test",
labels={},
lifecycle_state="ACTIVE",
)
}
provider.project_ids = project_ids
provider.default_project_id = GCP_PROJECT_ID
provider.identity = GCPIdentityInfo(
tests/providers/gcp/services/apikeys/apikeys_api_restrictions_configured/apikeys_api_restrictions_configured_test.py
+4 -4
View File
@@ -6,7 +6,7 @@ from tests.providers.gcp.gcp_fixtures import GCP_PROJECT_ID, set_mocked_gcp_prov
class Test_apikeys_api_restrictions_configured:
def test_apikeys_no_keys(self):
apikeys_client = mock.MagicMock
apikeys_client = mock.MagicMock()
apikeys_client.keys = []
with mock.patch(
@@ -40,7 +40,7 @@ class Test_apikeys_api_restrictions_configured:
project_id=GCP_PROJECT_ID,
)
apikeys_client = mock.MagicMock
apikeys_client = mock.MagicMock()
apikeys_client.project_ids = [GCP_PROJECT_ID]
apikeys_client.keys = [key]
apikeys_client.region = "global"
@@ -78,7 +78,7 @@ class Test_apikeys_api_restrictions_configured:
project_id=GCP_PROJECT_ID,
)
apikeys_client = mock.MagicMock
apikeys_client = mock.MagicMock()
apikeys_client.project_ids = [GCP_PROJECT_ID]
apikeys_client.keys = [key]
apikeys_client.region = "global"
@@ -122,7 +122,7 @@ class Test_apikeys_api_restrictions_configured:
project_id=GCP_PROJECT_ID,
)
apikeys_client = mock.MagicMock
apikeys_client = mock.MagicMock()
apikeys_client.project_ids = [GCP_PROJECT_ID]
apikeys_client.keys = [key]
apikeys_client.region = "global"
tests/providers/gcp/services/apikeys/apikeys_key_exists/apikeys_key_exists_test.py
+21 -2
View File
@@ -1,15 +1,25 @@
from re import search
from unittest import mock
from prowler.providers.gcp.models import GCPProject
from tests.providers.gcp.gcp_fixtures import GCP_PROJECT_ID, set_mocked_gcp_provider
class Test_apikeys_key_exists:
def test_apikeys_no_keys(self):
apikeys_client = mock.MagicMock
apikeys_client = mock.MagicMock()
apikeys_client.project_ids = [GCP_PROJECT_ID]
apikeys_client.keys = []
apikeys_client.region = "global"
apikeys_client.projects = {
GCP_PROJECT_ID: GCPProject(
id=GCP_PROJECT_ID,
number="123456789012",
name="test",
labels={},
lifecycle_state="ACTIVE",
)
}
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
@@ -43,10 +53,19 @@ class Test_apikeys_key_exists:
project_id=GCP_PROJECT_ID,
)
apikeys_client = mock.MagicMock
apikeys_client = mock.MagicMock()
apikeys_client.project_ids = [GCP_PROJECT_ID]
apikeys_client.keys = [key]
apikeys_client.region = "global"
apikeys_client.projects = {
GCP_PROJECT_ID: GCPProject(
id=GCP_PROJECT_ID,
number="123456789012",
name="test",
labels={},
lifecycle_state="ACTIVE",
)
}
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
tests/providers/gcp/services/apikeys/apikeys_key_rotated_in_90_days/apikeys_key_rotated_in_90_days_test.py
+3 -3
View File
@@ -7,7 +7,7 @@ from tests.providers.gcp.gcp_fixtures import GCP_PROJECT_ID, set_mocked_gcp_prov
class Test_apikeys_key_rotated_in_90_days:
def test_apikeys_no_keys(self):
apikeys_client = mock.MagicMock
apikeys_client = mock.MagicMock()
apikeys_client.keys = []
with mock.patch(
@@ -38,7 +38,7 @@ class Test_apikeys_key_rotated_in_90_days:
project_id=GCP_PROJECT_ID,
)
apikeys_client = mock.MagicMock
apikeys_client = mock.MagicMock()
apikeys_client.project_ids = [GCP_PROJECT_ID]
apikeys_client.keys = [key]
apikeys_client.region = "global"
@@ -78,7 +78,7 @@ class Test_apikeys_key_rotated_in_90_days:
project_id=GCP_PROJECT_ID,
)
apikeys_client = mock.MagicMock
apikeys_client = mock.MagicMock()
apikeys_client.project_ids = [GCP_PROJECT_ID]
apikeys_client.keys = [key]
apikeys_client.region = "global"
tests/providers/gcp/services/artifacts/artifacts_container_analysis_enabled/artifacts_container_analysis_enabled_test.py
+39 -14
View File
@@ -1,22 +1,35 @@
from unittest import mock
from prowler.providers.gcp.models import GCPProject
from prowler.providers.gcp.services.serviceusage.serviceusage_service import Service
from tests.providers.gcp.gcp_fixtures import GCP_PROJECT_ID, set_mocked_gcp_provider
class Test_artifacts_container_analysis_enabled:
def test_serviceusage_no_active_services(self):
serviceusage_client = mock.MagicMock
serviceusage_client = mock.MagicMock()
serviceusage_client.active_services = {}
serviceusage_client.project_ids = [GCP_PROJECT_ID]
serviceusage_client.region = "global"
serviceusage_client.projects = {
GCP_PROJECT_ID: GCPProject(
id=GCP_PROJECT_ID,
number="123456789012",
name="test",
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.artifacts.artifacts_container_analysis_enabled.artifacts_container_analysis_enabled.serviceusage_client",
new=serviceusage_client,
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_gcp_provider(),
),
mock.patch(
"prowler.providers.gcp.services.artifacts.artifacts_container_analysis_enabled.artifacts_container_analysis_enabled.serviceusage_client",
new=serviceusage_client,
),
):
from prowler.providers.gcp.services.artifacts.artifacts_container_analysis_enabled.artifacts_container_analysis_enabled import (
artifacts_container_analysis_enabled,
@@ -36,7 +49,7 @@ class Test_artifacts_container_analysis_enabled:
assert result[0].location == serviceusage_client.region
def test_serviceusage_active_cloudasset(self):
serviceusage_client = mock.MagicMock
serviceusage_client = mock.MagicMock()
serviceusage_client.active_services = {
GCP_PROJECT_ID: [
Service(
@@ -48,13 +61,25 @@ class Test_artifacts_container_analysis_enabled:
}
serviceusage_client.project_ids = [GCP_PROJECT_ID]
serviceusage_client.region = "global"
serviceusage_client.projects = {
GCP_PROJECT_ID: GCPProject(
id=GCP_PROJECT_ID,
number="123456789012",
name="test",
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.artifacts.artifacts_container_analysis_enabled.artifacts_container_analysis_enabled.serviceusage_client",
new=serviceusage_client,
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_gcp_provider(),
),
mock.patch(
"prowler.providers.gcp.services.artifacts.artifacts_container_analysis_enabled.artifacts_container_analysis_enabled.serviceusage_client",
new=serviceusage_client,
),
):
from prowler.providers.gcp.services.artifacts.artifacts_container_analysis_enabled.artifacts_container_analysis_enabled import (
artifacts_container_analysis_enabled,
tests/providers/gcp/services/bigquery/bigquery_dataset_cmk_encryption/bigquery_dataset_cmk_encryption_test.py
+3 -3
View File
@@ -5,7 +5,7 @@ from tests.providers.gcp.gcp_fixtures import GCP_PROJECT_ID, set_mocked_gcp_prov
class Test_bigquery_dataset_cmk_encryption:
def test_bigquery_no_datasets(self):
bigquery_client = mock.MagicMock
bigquery_client = mock.MagicMock()
bigquery_client.datasets = []
with mock.patch(
@@ -24,7 +24,7 @@ class Test_bigquery_dataset_cmk_encryption:
assert len(result) == 0
def test_one_compliant_dataset(self):
bigquery_client = mock.MagicMock
bigquery_client = mock.MagicMock()
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
@@ -64,7 +64,7 @@ class Test_bigquery_dataset_cmk_encryption:
assert result[0].location == dataset.region
def test_one_non_compliant_dataset(self):
bigquery_client = mock.MagicMock
bigquery_client = mock.MagicMock()
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
tests/providers/gcp/services/bigquery/bigquery_dataset_public_access/bigquery_dataset_public_access_test.py
+3 -3
View File
@@ -5,7 +5,7 @@ from tests.providers.gcp.gcp_fixtures import GCP_PROJECT_ID, set_mocked_gcp_prov
class Test_bigquery_dataset_public_access:
def test_bigquery_no_datasets(self):
bigquery_client = mock.MagicMock
bigquery_client = mock.MagicMock()
bigquery_client.datasets = []
with mock.patch(
@@ -35,7 +35,7 @@ class Test_bigquery_dataset_public_access:
project_id=GCP_PROJECT_ID,
)
bigquery_client = mock.MagicMock
bigquery_client = mock.MagicMock()
bigquery_client.project_ids = [GCP_PROJECT_ID]
bigquery_client.datasets = [dataset]
@@ -76,7 +76,7 @@ class Test_bigquery_dataset_public_access:
project_id=GCP_PROJECT_ID,
)
bigquery_client = mock.MagicMock
bigquery_client = mock.MagicMock()
bigquery_client.project_ids = [GCP_PROJECT_ID]
bigquery_client.datasets = [dataset]
tests/providers/gcp/services/bigquery/bigquery_table_cmk_encryption/bigquery_table_cmk_encryption_test.py
+3 -3
View File
@@ -5,7 +5,7 @@ from tests.providers.gcp.gcp_fixtures import GCP_PROJECT_ID, set_mocked_gcp_prov
class Test_bigquery_table_cmk_encryption:
def test_bigquery_no_tables(self):
bigquery_client = mock.MagicMock
bigquery_client = mock.MagicMock()
bigquery_client.tables = []
with mock.patch(
@@ -24,7 +24,7 @@ class Test_bigquery_table_cmk_encryption:
assert len(result) == 0
def test_one_compliant_table(self):
bigquery_client = mock.MagicMock
bigquery_client = mock.MagicMock()
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
@@ -64,7 +64,7 @@ class Test_bigquery_table_cmk_encryption:
assert result[0].location == table.region
def test_one_non_compliant_table(self):
bigquery_client = mock.MagicMock
bigquery_client = mock.MagicMock()
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
tests/providers/gcp/services/cloudresourcemanager/cloudresourcemanager_service_test.py
+5 -3
View File
@@ -26,9 +26,11 @@ class TestCloudResourceManagerService:
assert api_keys_client.service == "cloudresourcemanager"
assert api_keys_client.project_ids == [GCP_PROJECT_ID]
assert len(api_keys_client.projects) == 1
assert api_keys_client.projects[0].id == GCP_PROJECT_ID
assert api_keys_client.projects[0].audit_logging
assert len(api_keys_client.cloud_resource_manager_projects) == 1
assert (
api_keys_client.cloud_resource_manager_projects[0].id == GCP_PROJECT_ID
)
assert api_keys_client.cloud_resource_manager_projects[0].audit_logging
assert len(api_keys_client.bindings) == 2
assert (
tests/providers/gcp/services/cloudsql/cloudsql_instance_automated_backups/cloudsql_instance_automated_backups_test.py
+3 -3
View File
@@ -9,7 +9,7 @@ from tests.providers.gcp.gcp_fixtures import (
class Test_cloudsql_instance_automated_backups:
def test_no_cloudsql_instances(self):
cloudsql_client = mock.MagicMock
cloudsql_client = mock.MagicMock()
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
@@ -29,7 +29,7 @@ class Test_cloudsql_instance_automated_backups:
assert len(result) == 0
def test_cloudsql_instance_with_automated_backups(self):
cloudsql_client = mock.MagicMock
cloudsql_client = mock.MagicMock()
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
@@ -75,7 +75,7 @@ class Test_cloudsql_instance_automated_backups:
assert result[0].project_id == GCP_PROJECT_ID
def test_cloudsql_instance_without_automated_backups(self):
cloudsql_client = mock.MagicMock
cloudsql_client = mock.MagicMock()
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
tests/providers/gcp/services/cloudsql/cloudsql_instance_mysql_local_infile_flag/cloudsql_instance_mysql_local_infile_flag_test.py
+5 -5
View File
@@ -9,7 +9,7 @@ from tests.providers.gcp.gcp_fixtures import (
class Test_cloudsql_instance_mysql_local_infile_flag:
def test_no_cloudsql_instances(self):
cloudsql_client = mock.MagicMock
cloudsql_client = mock.MagicMock()
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
@@ -29,7 +29,7 @@ class Test_cloudsql_instance_mysql_local_infile_flag:
assert len(result) == 0
def test_cloudsql_postgres_instance(self):
cloudsql_client = mock.MagicMock
cloudsql_client = mock.MagicMock()
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
@@ -66,7 +66,7 @@ class Test_cloudsql_instance_mysql_local_infile_flag:
assert len(result) == 0
def test_cloudsql_instance_with_no_flags(self):
cloudsql_client = mock.MagicMock
cloudsql_client = mock.MagicMock()
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
@@ -112,7 +112,7 @@ class Test_cloudsql_instance_mysql_local_infile_flag:
assert result[0].project_id == GCP_PROJECT_ID
def test_cloudsql_instance_with_local_infile_flag_off(self):
cloudsql_client = mock.MagicMock
cloudsql_client = mock.MagicMock()
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
@@ -158,7 +158,7 @@ class Test_cloudsql_instance_mysql_local_infile_flag:
assert result[0].project_id == GCP_PROJECT_ID
def test_cloudsql_instance_with_local_infile_flag_on(self):
cloudsql_client = mock.MagicMock
cloudsql_client = mock.MagicMock()
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
tests/providers/gcp/services/cloudsql/cloudsql_instance_mysql_skip_show_database_flag/cloudsql_instance_mysql_skip_show_database_flag_test.py
+5 -5
View File
@@ -9,7 +9,7 @@ from tests.providers.gcp.gcp_fixtures import (
class Test_cloudsql_instance_mysql_skip_show_database_flag:
def test_no_cloudsql_instances(self):
cloudsql_client = mock.MagicMock
cloudsql_client = mock.MagicMock()
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
@@ -29,7 +29,7 @@ class Test_cloudsql_instance_mysql_skip_show_database_flag:
assert len(result) == 0
def test_cloudsql_postgres_instance(self):
cloudsql_client = mock.MagicMock
cloudsql_client = mock.MagicMock()
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
@@ -66,7 +66,7 @@ class Test_cloudsql_instance_mysql_skip_show_database_flag:
assert len(result) == 0
def test_cloudsql_instance_no_flags(self):
cloudsql_client = mock.MagicMock
cloudsql_client = mock.MagicMock()
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
@@ -112,7 +112,7 @@ class Test_cloudsql_instance_mysql_skip_show_database_flag:
assert result[0].project_id == GCP_PROJECT_ID
def test_cloudsql_instance_with_skip_show_databases_off(self):
cloudsql_client = mock.MagicMock
cloudsql_client = mock.MagicMock()
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
@@ -158,7 +158,7 @@ class Test_cloudsql_instance_mysql_skip_show_database_flag:
assert result[0].project_id == GCP_PROJECT_ID
def test_cloudsql_instance_with_skip_show_databases_on(self):
cloudsql_client = mock.MagicMock
cloudsql_client = mock.MagicMock()
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
tests/providers/gcp/services/cloudsql/cloudsql_instance_postgres_enable_pgaudit_flag/cloudsql_instance_postgres_enable_pgaudit_flag_test.py
+5 -5
View File
@@ -9,7 +9,7 @@ from tests.providers.gcp.gcp_fixtures import (
class Test_cloudsql_instance_postgres_enable_pgaudit_flag:
def test_no_cloudsql_instances(self):
cloudsql_client = mock.MagicMock
cloudsql_client = mock.MagicMock()
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
@@ -29,7 +29,7 @@ class Test_cloudsql_instance_postgres_enable_pgaudit_flag:
assert len(result) == 0
def test_cloudsql_mysql_instance(self):
cloudsql_client = mock.MagicMock
cloudsql_client = mock.MagicMock()
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
@@ -66,7 +66,7 @@ class Test_cloudsql_instance_postgres_enable_pgaudit_flag:
assert len(result) == 0
def test_cloudsql_instance_no_flags(self):
cloudsql_client = mock.MagicMock
cloudsql_client = mock.MagicMock()
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
@@ -112,7 +112,7 @@ class Test_cloudsql_instance_postgres_enable_pgaudit_flag:
assert result[0].project_id == GCP_PROJECT_ID
def test_cloudsql_instance_pgaudit_flag_off(self):
cloudsql_client = mock.MagicMock
cloudsql_client = mock.MagicMock()
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
@@ -158,7 +158,7 @@ class Test_cloudsql_instance_postgres_enable_pgaudit_flag:
assert result[0].project_id == GCP_PROJECT_ID
def test_cloudsql_instance_pgaudit_flag_on(self):
cloudsql_client = mock.MagicMock
cloudsql_client = mock.MagicMock()
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
tests/providers/gcp/services/cloudsql/cloudsql_instance_postgres_log_connections_flag/cloudsql_instance_postgres_log_connections_flag_test.py
+5 -5
View File
@@ -9,7 +9,7 @@ from tests.providers.gcp.gcp_fixtures import (
class Test_cloudsql_instance_postgres_log_connections_flag:
def test_no_cloudsql_instances(self):
cloudsql_client = mock.MagicMock
cloudsql_client = mock.MagicMock()
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
@@ -29,7 +29,7 @@ class Test_cloudsql_instance_postgres_log_connections_flag:
assert len(result) == 0
def test_cloudsql_mysql_instance(self):
cloudsql_client = mock.MagicMock
cloudsql_client = mock.MagicMock()
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
@@ -66,7 +66,7 @@ class Test_cloudsql_instance_postgres_log_connections_flag:
assert len(result) == 0
def test_cloudsql_instance_no_flags(self):
cloudsql_client = mock.MagicMock
cloudsql_client = mock.MagicMock()
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
@@ -112,7 +112,7 @@ class Test_cloudsql_instance_postgres_log_connections_flag:
assert result[0].project_id == GCP_PROJECT_ID
def test_cloudsql_instance_log_connections_flag_off(self):
cloudsql_client = mock.MagicMock
cloudsql_client = mock.MagicMock()
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
@@ -158,7 +158,7 @@ class Test_cloudsql_instance_postgres_log_connections_flag:
assert result[0].project_id == GCP_PROJECT_ID
def test_cloudsql_instance_log_connections_flag_on(self):
cloudsql_client = mock.MagicMock
cloudsql_client = mock.MagicMock()
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
tests/providers/gcp/services/cloudsql/cloudsql_instance_postgres_log_disconnections_flag/cloudsql_instance_postgres_log_disconnections_flag_test.py
+5 -5
View File
@@ -9,7 +9,7 @@ from tests.providers.gcp.gcp_fixtures import (
class Test_cloudsql_instance_postgres_log_disconnections_flag:
def test_no_cloudsql_instances(self):
cloudsql_client = mock.MagicMock
cloudsql_client = mock.MagicMock()
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
@@ -29,7 +29,7 @@ class Test_cloudsql_instance_postgres_log_disconnections_flag:
assert len(result) == 0
def test_cloudsql_mysql_instance(self):
cloudsql_client = mock.MagicMock
cloudsql_client = mock.MagicMock()
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
@@ -66,7 +66,7 @@ class Test_cloudsql_instance_postgres_log_disconnections_flag:
assert len(result) == 0
def test_cloudsql_instance_no_flags(self):
cloudsql_client = mock.MagicMock
cloudsql_client = mock.MagicMock()
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
@@ -112,7 +112,7 @@ class Test_cloudsql_instance_postgres_log_disconnections_flag:
assert result[0].project_id == GCP_PROJECT_ID
def test_cloudsql_instance_log_disconnections_flag_off(self):
cloudsql_client = mock.MagicMock
cloudsql_client = mock.MagicMock()
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
@@ -158,7 +158,7 @@ class Test_cloudsql_instance_postgres_log_disconnections_flag:
assert result[0].project_id == GCP_PROJECT_ID
def test_cloudsql_instance_log_disconnections_flag_on(self):
cloudsql_client = mock.MagicMock
cloudsql_client = mock.MagicMock()
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
tests/providers/gcp/services/cloudsql/cloudsql_instance_postgres_log_error_verbosity_flag/cloudsql_instance_postgres_log_error_verbosity_flag_test.py
+5 -5
View File
@@ -9,7 +9,7 @@ from tests.providers.gcp.gcp_fixtures import (
class Test_cloudsql_instance_postgres_log_error_verbosity_flag:
def test_no_cloudsql_instances(self):
cloudsql_client = mock.MagicMock
cloudsql_client = mock.MagicMock()
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
@@ -29,7 +29,7 @@ class Test_cloudsql_instance_postgres_log_error_verbosity_flag:
assert len(result) == 0
def test_cloudsql_mysql_instance(self):
cloudsql_client = mock.MagicMock
cloudsql_client = mock.MagicMock()
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
@@ -66,7 +66,7 @@ class Test_cloudsql_instance_postgres_log_error_verbosity_flag:
assert len(result) == 0
def test_cloudsql_instance_no_flags(self):
cloudsql_client = mock.MagicMock
cloudsql_client = mock.MagicMock()
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
@@ -112,7 +112,7 @@ class Test_cloudsql_instance_postgres_log_error_verbosity_flag:
assert result[0].project_id == GCP_PROJECT_ID
def test_cloudsql_instance_log_error_verbosity_flag_off(self):
cloudsql_client = mock.MagicMock
cloudsql_client = mock.MagicMock()
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
@@ -158,7 +158,7 @@ class Test_cloudsql_instance_postgres_log_error_verbosity_flag:
assert result[0].project_id == GCP_PROJECT_ID
def test_cloudsql_instance_log_error_verbosity_flag_on(self):
cloudsql_client = mock.MagicMock
cloudsql_client = mock.MagicMock()
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
tests/providers/gcp/services/cloudsql/cloudsql_instance_postgres_log_min_duration_statement_flag/cloudsql_instance_postgres_log_min_duration_statement_flag_test.py
+5 -5
View File
@@ -9,7 +9,7 @@ from tests.providers.gcp.gcp_fixtures import (
class Test_cloudsql_instance_postgres_log_min_duration_statement_flag:
def test_no_cloudsql_instances(self):
cloudsql_client = mock.MagicMock
cloudsql_client = mock.MagicMock()
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
@@ -29,7 +29,7 @@ class Test_cloudsql_instance_postgres_log_min_duration_statement_flag:
assert len(result) == 0
def test_cloudsql_mysql_instance(self):
cloudsql_client = mock.MagicMock
cloudsql_client = mock.MagicMock()
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
@@ -66,7 +66,7 @@ class Test_cloudsql_instance_postgres_log_min_duration_statement_flag:
assert len(result) == 0
def test_cloudsql_instance_no_flags(self):
cloudsql_client = mock.MagicMock
cloudsql_client = mock.MagicMock()
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
@@ -112,7 +112,7 @@ class Test_cloudsql_instance_postgres_log_min_duration_statement_flag:
assert result[0].project_id == GCP_PROJECT_ID
def test_cloudsql_instance_log_min_duration_statement_flag_off(self):
cloudsql_client = mock.MagicMock
cloudsql_client = mock.MagicMock()
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
@@ -158,7 +158,7 @@ class Test_cloudsql_instance_postgres_log_min_duration_statement_flag:
assert result[0].project_id == GCP_PROJECT_ID
def test_cloudsql_instance_log_min_duration_statement_flag_on(self):
cloudsql_client = mock.MagicMock
cloudsql_client = mock.MagicMock()
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
tests/providers/gcp/services/cloudsql/cloudsql_instance_postgres_log_min_error_statement_flag/cloudsql_instance_postgres_log_min_error_statement_flag_test.py
+5 -5
View File
@@ -9,7 +9,7 @@ from tests.providers.gcp.gcp_fixtures import (
class Test_cloudsql_instance_postgres_log_min_error_statement_flag:
def test_no_cloudsql_instances(self):
cloudsql_client = mock.MagicMock
cloudsql_client = mock.MagicMock()
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
@@ -29,7 +29,7 @@ class Test_cloudsql_instance_postgres_log_min_error_statement_flag:
assert len(result) == 0
def test_cloudsql_mysql_instance(self):
cloudsql_client = mock.MagicMock
cloudsql_client = mock.MagicMock()
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
@@ -66,7 +66,7 @@ class Test_cloudsql_instance_postgres_log_min_error_statement_flag:
assert len(result) == 0
def test_cloudsql_instance_no_flags(self):
cloudsql_client = mock.MagicMock
cloudsql_client = mock.MagicMock()
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
@@ -112,7 +112,7 @@ class Test_cloudsql_instance_postgres_log_min_error_statement_flag:
assert result[0].project_id == GCP_PROJECT_ID
def test_cloudsql_instance_log_min_error_statement_flag_off(self):
cloudsql_client = mock.MagicMock
cloudsql_client = mock.MagicMock()
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
@@ -158,7 +158,7 @@ class Test_cloudsql_instance_postgres_log_min_error_statement_flag:
assert result[0].project_id == GCP_PROJECT_ID
def test_cloudsql_instance_log_min_error_statement_flag_on(self):
cloudsql_client = mock.MagicMock
cloudsql_client = mock.MagicMock()
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
tests/providers/gcp/services/cloudsql/cloudsql_instance_postgres_log_min_messages_flag/cloudsql_instance_postgres_log_min_messages_flag_test.py
+5 -5
View File
@@ -9,7 +9,7 @@ from tests.providers.gcp.gcp_fixtures import (
class Test_cloudsql_instance_postgres_log_min_messages_flag:
def test_no_cloudsql_instances(self):
cloudsql_client = mock.MagicMock
cloudsql_client = mock.MagicMock()
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
@@ -29,7 +29,7 @@ class Test_cloudsql_instance_postgres_log_min_messages_flag:
assert len(result) == 0
def test_cloudsql_mysql_instance(self):
cloudsql_client = mock.MagicMock
cloudsql_client = mock.MagicMock()
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
@@ -66,7 +66,7 @@ class Test_cloudsql_instance_postgres_log_min_messages_flag:
assert len(result) == 0
def test_cloudsql_instance_no_flags(self):
cloudsql_client = mock.MagicMock
cloudsql_client = mock.MagicMock()
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
@@ -112,7 +112,7 @@ class Test_cloudsql_instance_postgres_log_min_messages_flag:
assert result[0].project_id == GCP_PROJECT_ID
def test_cloudsql_instance_log_min_messages_flag_off(self):
cloudsql_client = mock.MagicMock
cloudsql_client = mock.MagicMock()
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
@@ -158,7 +158,7 @@ class Test_cloudsql_instance_postgres_log_min_messages_flag:
assert result[0].project_id == GCP_PROJECT_ID
def test_cloudsql_instance_log_min_messages_flag_on(self):
cloudsql_client = mock.MagicMock
cloudsql_client = mock.MagicMock()
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",

Some files were not shown because too many files have changed in this diff Show More