feat(gcp): add cloudsql_instance_high_availability_enabled check (#11024)

Co-authored-by: Lydia Vilchez <lydiavilchezlopez@gmail.com>
Co-authored-by: lydiavilchez <114735608+lydiavilchez@users.noreply.github.com>
This commit is contained in:
authored and GitHub committed 2026-06-12 11:51:13 +02:00
1 parent 56831a7392
commit 79e066d3f5
9 files changed
+291

No files matched your search

+1
View File
@@ -9,6 +9,7 @@ All notable changes to the **Prowler SDK** are documented in this file.
- `securityhub_delegated_admin_enabled_all_regions` check for AWS provider, verifying that Security Hub has a delegated administrator, is active in all opted-in regions, and has organization auto-enable on [(#11259)](https://github.com/prowler-cloud/prowler/pull/11259)
- `config_delegated_admin_and_org_aggregator_all_regions` check for AWS provider, verifying that AWS Config has a delegated administrator and an organization aggregator covering all AWS regions [(#11259)](https://github.com/prowler-cloud/prowler/pull/11259)
- `sagemaker_clarify_exists` check for AWS provider [(#11211)](https://github.com/prowler-cloud/prowler/pull/11211)
- `cloudsql_instance_high_availability_enabled` check for GCP provider, verifying Cloud SQL primary instances use `REGIONAL` availability for automatic zone failover [(#11024)](https://github.com/prowler-cloud/prowler/pull/11024)
- `identity_storage_service_level_admins_scoped` check for OCI provider CIS 3.1 control 1.15, ensuring storage service-level administrators exclude delete permissions [(#11523)](https://github.com/prowler-cloud/prowler/pull/11523)
---
@@ -0,0 +1,38 @@
{
"Provider": "gcp",
"CheckID": "cloudsql_instance_high_availability_enabled",
"CheckTitle": "Cloud SQL instance has high availability (REGIONAL) configured",
"CheckType": [],
"ServiceName": "cloudsql",
"SubServiceName": "",
"ResourceIdTemplate": "",
"Severity": "medium",
"ResourceType": "sqladmin.googleapis.com/Instance",
"Description": "Ensures that Cloud SQL instances have high availability configured by setting availabilityType to REGIONAL. A REGIONAL instance maintains a standby replica in a different zone within the same region and automatically fails over on zone-level outages.",
"Risk": "Instances with ZONAL availability have no standby replica. A zone-level outage will cause database downtime until manual recovery, violating availability requirements and potentially breaching SLAs and ISMS-P 2.12.1 disaster preparedness controls.",
"RelatedUrl": "",
"AdditionalURLs": [
"https://cloud.google.com/sql/docs/postgres/high-availability",
"https://cloud.google.com/sql/docs/sqlserver/high-availability"
],
"Remediation": {
"Code": {
"CLI": "gcloud sql instances patch <INSTANCE_NAME> --availability-type=REGIONAL",
"NativeIaC": "",
"Other": "1. Go to Google Cloud Console > SQL > Instances.\n2. Click the instance name, then Edit.\n3. Under Availability, select Multiple zones (Highly available).\n4. Click Save.",
"Terraform": "```hcl\nresource \"google_sql_database_instance\" \"example\" {\n name = \"<instance_name>\"\n database_version = \"POSTGRES_15\"\n region = \"<region>\"\n\n settings {\n tier = \"db-custom-2-7680\"\n\n availability_type = \"REGIONAL\" # Critical: enables HA standby replica\n\n backup_configuration {\n enabled = true\n start_time = \"02:00\"\n }\n }\n}\n```"
},
"Recommendation": {
"Text": "Set availabilityType to REGIONAL for all production Cloud SQL instances. This creates a standby replica in a different zone and enables automatic failover, reducing RTO in the event of a zone outage.",
"Url": "https://hub.prowler.com/check/cloudsql_instance_high_availability_enabled"
}
},
"Categories": [
"resilience"
],
"DependsOn": [],
"RelatedTo": [
"cloudsql_instance_automated_backups"
],
"Notes": "Enabling HA increases instance cost approximately 2x due to the standby replica. ZONAL instances are acceptable for non-production workloads where downtime is tolerable."
}
@@ -0,0 +1,41 @@
from prowler.lib.check.models import Check, Check_Report_GCP
from prowler.providers.gcp.services.cloudsql.cloudsql_client import cloudsql_client
class cloudsql_instance_high_availability_enabled(Check):
"""Check that Cloud SQL primary instances are configured for high availability.
Verifies that each Cloud SQL primary instance has `availabilityType` set to
`REGIONAL`, which provisions a standby replica in a different zone within
the same region and enables automatic failover on zone-level outages. Read
replicas are skipped because they inherit availability from their primary.
"""
def execute(self) -> list[Check_Report_GCP]:
"""Execute the high availability check across all Cloud SQL instances.
Returns:
A list of `Check_Report_GCP` findings, one per Cloud SQL primary
instance. Status is `PASS` when `availability_type == "REGIONAL"`
and `FAIL` otherwise.
"""
findings = []
for instance in cloudsql_client.instances:
if instance.instance_type != "CLOUD_SQL_INSTANCE":
continue
report = Check_Report_GCP(metadata=self.metadata(), resource=instance)
if instance.availability_type == "REGIONAL":
report.status = "PASS"
report.status_extended = (
f"Database instance {instance.name} has high availability "
f"(REGIONAL) configured."
)
else:
report.status = "FAIL"
report.status_extended = (
f"Database instance {instance.name} does not have high "
f"availability configured (current: "
f"{instance.availability_type})."
)
findings.append(report)
return findings
@@ -46,6 +46,9 @@ class CloudSQL(GCPService):
"authorizedNetworks", []
),
flags=settings.get("databaseFlags", []),
availability_type=settings.get(
"availabilityType", "ZONAL"
),
instance_type=instance.get(
"instanceType", "CLOUD_SQL_INSTANCE"
),
@@ -76,6 +79,7 @@ class Instance(BaseModel):
ssl_mode: str
automated_backups: bool
flags: list
availability_type: str = "ZONAL"
instance_type: str = "CLOUD_SQL_INSTANCE"
cmek_key_name: Optional[str] = None
project_id: str
+2
View File
@@ -722,6 +722,7 @@ def mock_api_instances_calls(client: MagicMock, service: str):
},
"backupConfiguration": {"enabled": True},
"databaseFlags": [],
"availabilityType": "REGIONAL",
},
},
{
@@ -737,6 +738,7 @@ def mock_api_instances_calls(client: MagicMock, service: str):
},
"backupConfiguration": {"enabled": False},
"databaseFlags": [],
"availabilityType": "ZONAL",
},
},
]
@@ -0,0 +1,205 @@
from unittest import mock
from tests.providers.gcp.gcp_fixtures import (
GCP_EU1_LOCATION,
GCP_PROJECT_ID,
set_mocked_gcp_provider,
)
class Test_cloudsql_instance_high_availability_enabled:
"""Tests for the cloudsql_instance_high_availability_enabled check."""
def test_no_instances(self):
"""No Cloud SQL instances → no findings."""
cloudsql_client = mock.MagicMock()
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_gcp_provider(),
),
mock.patch(
"prowler.providers.gcp.services.cloudsql.cloudsql_instance_high_availability_enabled.cloudsql_instance_high_availability_enabled.cloudsql_client",
new=cloudsql_client,
),
):
from prowler.providers.gcp.services.cloudsql.cloudsql_instance_high_availability_enabled.cloudsql_instance_high_availability_enabled import (
cloudsql_instance_high_availability_enabled,
)
cloudsql_client.instances = []
check = cloudsql_instance_high_availability_enabled()
result = check.execute()
assert len(result) == 0
def test_instance_ha_enabled(self):
"""A REGIONAL primary instance → PASS."""
cloudsql_client = mock.MagicMock()
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_gcp_provider(),
),
mock.patch(
"prowler.providers.gcp.services.cloudsql.cloudsql_instance_high_availability_enabled.cloudsql_instance_high_availability_enabled.cloudsql_client",
new=cloudsql_client,
),
):
from prowler.providers.gcp.services.cloudsql.cloudsql_instance_high_availability_enabled.cloudsql_instance_high_availability_enabled import (
cloudsql_instance_high_availability_enabled,
)
from prowler.providers.gcp.services.cloudsql.cloudsql_service import (
Instance,
)
cloudsql_client.instances = [
Instance(
name="db-ha",
version="POSTGRES_15",
ip_addresses=[],
region=GCP_EU1_LOCATION,
public_ip=False,
require_ssl=False,
ssl_mode="ENCRYPTED_ONLY",
automated_backups=True,
authorized_networks=[],
flags=[],
project_id=GCP_PROJECT_ID,
availability_type="REGIONAL",
)
]
check = cloudsql_instance_high_availability_enabled()
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert result[0].resource_id == "db-ha"
assert result[0].location == GCP_EU1_LOCATION
assert result[0].project_id == GCP_PROJECT_ID
def test_instance_ha_disabled(self):
"""A ZONAL primary instance → FAIL with current availability in status_extended."""
cloudsql_client = mock.MagicMock()
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_gcp_provider(),
),
mock.patch(
"prowler.providers.gcp.services.cloudsql.cloudsql_instance_high_availability_enabled.cloudsql_instance_high_availability_enabled.cloudsql_client",
new=cloudsql_client,
),
):
from prowler.providers.gcp.services.cloudsql.cloudsql_instance_high_availability_enabled.cloudsql_instance_high_availability_enabled import (
cloudsql_instance_high_availability_enabled,
)
from prowler.providers.gcp.services.cloudsql.cloudsql_service import (
Instance,
)
cloudsql_client.instances = [
Instance(
name="db-zonal",
version="POSTGRES_15",
ip_addresses=[],
region=GCP_EU1_LOCATION,
public_ip=False,
require_ssl=False,
ssl_mode="ENCRYPTED_ONLY",
automated_backups=True,
authorized_networks=[],
flags=[],
project_id=GCP_PROJECT_ID,
availability_type="ZONAL",
)
]
check = cloudsql_instance_high_availability_enabled()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert "ZONAL" in result[0].status_extended
assert result[0].resource_id == "db-zonal"
assert result[0].location == GCP_EU1_LOCATION
assert result[0].project_id == GCP_PROJECT_ID
def test_read_replica_skipped(self):
"""Read replicas (instance_type != CLOUD_SQL_INSTANCE) are skipped."""
cloudsql_client = mock.MagicMock()
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_gcp_provider(),
),
mock.patch(
"prowler.providers.gcp.services.cloudsql.cloudsql_instance_high_availability_enabled.cloudsql_instance_high_availability_enabled.cloudsql_client",
new=cloudsql_client,
),
):
from prowler.providers.gcp.services.cloudsql.cloudsql_instance_high_availability_enabled.cloudsql_instance_high_availability_enabled import (
cloudsql_instance_high_availability_enabled,
)
from prowler.providers.gcp.services.cloudsql.cloudsql_service import (
Instance,
)
cloudsql_client.instances = [
Instance(
name="db-replica",
version="POSTGRES_15",
ip_addresses=[],
region=GCP_EU1_LOCATION,
public_ip=False,
require_ssl=False,
ssl_mode="ENCRYPTED_ONLY",
automated_backups=True,
authorized_networks=[],
flags=[],
project_id=GCP_PROJECT_ID,
availability_type="ZONAL",
instance_type="READ_REPLICA_INSTANCE",
)
]
check = cloudsql_instance_high_availability_enabled()
result = check.execute()
assert len(result) == 0
def test_instance_default_availability_type_fails(self):
"""An instance missing availabilityType defaults to ZONAL (service layer) and must FAIL."""
cloudsql_client = mock.MagicMock()
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_gcp_provider(),
),
mock.patch(
"prowler.providers.gcp.services.cloudsql.cloudsql_instance_high_availability_enabled.cloudsql_instance_high_availability_enabled.cloudsql_client",
new=cloudsql_client,
),
):
from prowler.providers.gcp.services.cloudsql.cloudsql_instance_high_availability_enabled.cloudsql_instance_high_availability_enabled import (
cloudsql_instance_high_availability_enabled,
)
from prowler.providers.gcp.services.cloudsql.cloudsql_service import (
Instance,
)
cloudsql_client.instances = [
Instance(
name="db-default",
version="POSTGRES_15",
ip_addresses=[],
region=GCP_EU1_LOCATION,
public_ip=False,
require_ssl=False,
ssl_mode="ENCRYPTED_ONLY",
automated_backups=True,
authorized_networks=[],
flags=[],
project_id=GCP_PROJECT_ID,
# availability_type omitted → model default "ZONAL"
)
]
check = cloudsql_instance_high_availability_enabled()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert "ZONAL" in result[0].status_extended