feat(gcp): add compute_snapshot_not_outdated check (#9774)

This commit is contained in:
lydiavilchez
2026-01-14 12:35:29 +01:00
committed by GitHub
parent 9844379d30
commit 2cde4c939d
9 changed files with 501 additions and 0 deletions
+1
View File
@@ -18,6 +18,7 @@ All notable changes to the **Prowler SDK** are documented in this file.
- `Cloudflare` provider with critical security checks [(#9423)](https://github.com/prowler-cloud/prowler/pull/9423)
- `compute_instance_single_network_interface` check for GCP provider [(#9702)](https://github.com/prowler-cloud/prowler/pull/9702)
- `compute_image_not_publicly_shared` check for GCP provider [(#9718)](https://github.com/prowler-cloud/prowler/pull/9718)
- `compute_snapshot_not_outdated` check for GCP provider [(#9774)](https://github.com/prowler-cloud/prowler/pull/9774)
- CIS 1.12 compliance framework for Kubernetes [(#9778)](https://github.com/prowler-cloud/prowler/pull/9778)
- CIS 6.0 for M365 provider [(#9779)](https://github.com/prowler-cloud/prowler/pull/9779)
- CIS 5.0 compliance framework for the Azure provider [(#9777)](https://github.com/prowler-cloud/prowler/pull/9777)
+3
View File
@@ -510,6 +510,9 @@ gcp:
# gcp.compute_instance_group_multiple_zones
# Minimum number of zones a MIG should span for high availability
mig_min_zones: 2
# gcp.compute_snapshot_not_outdated
# Maximum age in days for disk snapshots before they are considered outdated
max_snapshot_age_days: 90
# GCP Service Account and user-managed keys unused configuration
# gcp.iam_service_account_unused
# gcp.iam_sa_user_managed_key_unused
@@ -1,3 +1,4 @@
from datetime import datetime
from typing import Optional
from pydantic.v1 import BaseModel
@@ -22,6 +23,7 @@ class Compute(GCPService):
self.load_balancers = []
self.instance_groups = []
self.images = []
self.snapshots = []
self._get_regions()
self._get_projects()
self._get_url_maps()
@@ -36,6 +38,7 @@ class Compute(GCPService):
self.__threading_call__(self._get_zonal_instance_groups, self.zones)
self._associate_migs_with_load_balancers()
self._get_images()
self._get_snapshots()
def _get_regions(self):
for project_id in self.project_ids:
@@ -602,6 +605,57 @@ class Compute(GCPService):
f"{project_id} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
def _get_snapshots(self) -> None:
for project_id in self.project_ids:
try:
request = self.client.snapshots().list(project=project_id)
while request is not None:
response = request.execute(num_retries=DEFAULT_RETRY_ATTEMPTS)
for snapshot in response.get("items", []):
# Parse creation timestamp to datetime
creation_timestamp_str = snapshot.get("creationTimestamp", "")
creation_timestamp = None
if creation_timestamp_str:
try:
# GCP timestamps are in RFC 3339 format
creation_timestamp = datetime.fromisoformat(
creation_timestamp_str.replace("Z", "+00:00")
)
except ValueError:
logger.error(
f"Could not parse timestamp {creation_timestamp_str} for snapshot {snapshot['name']}"
)
# Extract source disk name from the full URL
source_disk_url = snapshot.get("sourceDisk", "")
source_disk = (
source_disk_url.split("/")[-1] if source_disk_url else ""
)
self.snapshots.append(
Snapshot(
name=snapshot["name"],
id=snapshot["id"],
project_id=project_id,
creation_timestamp=creation_timestamp,
source_disk=source_disk,
source_disk_id=snapshot.get("sourceDiskId"),
disk_size_gb=int(snapshot.get("diskSizeGb", 0)),
storage_bytes=int(snapshot.get("storageBytes", 0)),
storage_locations=snapshot.get("storageLocations", []),
status=snapshot.get("status", ""),
auto_created=snapshot.get("autoCreated", False),
)
)
request = self.client.snapshots().list_next(
previous_request=request, previous_response=response
)
except Exception as error:
logger.error(
f"{project_id} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
class NetworkInterface(BaseModel):
name: str
@@ -708,3 +762,17 @@ class Image(BaseModel):
id: str
project_id: str
publicly_shared: bool = False
class Snapshot(BaseModel):
name: str
id: str
project_id: str
creation_timestamp: Optional[datetime] = None
source_disk: str = ""
source_disk_id: Optional[str] = None
disk_size_gb: int = 0
storage_bytes: int = 0
storage_locations: list[str] = []
status: str = ""
auto_created: bool = False
@@ -0,0 +1,38 @@
{
"Provider": "gcp",
"CheckID": "compute_snapshot_not_outdated",
"CheckTitle": "Compute Engine disk snapshot is not outdated",
"CheckType": [],
"ServiceName": "compute",
"SubServiceName": "",
"ResourceIdTemplate": "",
"Severity": "low",
"ResourceType": "compute.googleapis.com/Snapshot",
"ResourceGroup": "storage",
"Description": "Compute Engine **disk snapshots** are evaluated against a configurable age threshold (default `90` days) to identify snapshots exceeding the organization's retention policy.",
"Risk": "Outdated snapshots containing **sensitive data** expand the **attack surface** and risk data exposure if compromised.\n\nStale snapshots may violate compliance requirements, complicate disaster recovery efforts, and introduce configuration drift that affects system **integrity**.",
"RelatedUrl": "",
"AdditionalURLs": [
"https://www.trendmicro.com/cloudoneconformity/knowledge-base/gcp/ComputeEngine/remove-old-disk-snapshots.html",
"https://cloud.google.com/compute/docs/disks/create-snapshots",
"https://cloud.google.com/compute/docs/disks/snapshot-best-practices"
],
"Remediation": {
"Code": {
"CLI": "gcloud compute snapshots delete SNAPSHOT_NAME --project=PROJECT_ID",
"NativeIaC": "",
"Other": "1. Open Google Cloud Console and navigate to Compute Engine > Snapshots\n2. Identify snapshots older than your retention policy\n3. Select outdated snapshots and click **Delete**\n4. Confirm the deletion\n\nTo automate cleanup, create a snapshot schedule with auto-delete policies under Compute Engine > Snapshots > Snapshot schedules.",
"Terraform": "```hcl\nresource \"google_compute_resource_policy\" \"snapshot_schedule\" {\n name = \"snapshot-schedule-with-retention\"\n region = var.region\n\n snapshot_schedule_policy {\n schedule {\n daily_schedule {\n days_in_cycle = 1\n start_time = \"04:00\"\n }\n }\n\n # Automatically delete snapshots older than 90 days\n retention_policy {\n max_retention_days = 90\n on_source_disk_delete = \"KEEP_AUTO_SNAPSHOTS\"\n }\n }\n}\n\nresource \"google_compute_disk_resource_policy_attachment\" \"attachment\" {\n name = google_compute_resource_policy.snapshot_schedule.name\n disk = google_compute_disk.example.name\n zone = var.zone\n}\n```"
},
"Recommendation": {
"Text": "Implement a snapshot lifecycle policy to automatically delete snapshots older than your organization's retention requirements. Regularly review and clean up outdated snapshots to reduce storage costs and minimize data exposure risks. Consider using scheduled snapshots with automatic deletion policies.",
"Url": "https://hub.prowler.com/check/compute_snapshot_not_outdated"
}
},
"Categories": [
"resilience"
],
"DependsOn": [],
"RelatedTo": [],
"Notes": "The age threshold is configurable via the `max_snapshot_age_days` parameter in the configuration file (default: 90 days). Snapshots without a creation timestamp will be flagged for manual review."
}
@@ -0,0 +1,60 @@
from datetime import datetime, timezone
from prowler.lib.check.models import Check, Check_Report_GCP
from prowler.providers.gcp.services.compute.compute_client import compute_client
class compute_snapshot_not_outdated(Check):
"""Check that Compute Engine disk snapshots are not outdated.
This check ensures Compute Engine disk snapshots are within the configured
age threshold (default 90 days) to help control storage costs and limit
exposure from stale data.
- PASS: Snapshot is not outdated (within the acceptable age threshold).
- FAIL: Snapshot is outdated (exceeds the configured age threshold).
"""
def execute(self) -> list[Check_Report_GCP]:
findings = []
max_snapshot_age_days = compute_client.audit_config.get(
"max_snapshot_age_days", 90
)
current_time = datetime.now(timezone.utc)
for snapshot in compute_client.snapshots:
report = Check_Report_GCP(
metadata=self.metadata(),
resource=snapshot,
location="global",
)
if snapshot.creation_timestamp is None:
report.status = "FAIL"
report.status_extended = (
f"Disk snapshot {snapshot.name} timestamp could not be retrieved "
"and cannot be evaluated for age."
)
findings.append(report)
continue
snapshot_age = (current_time - snapshot.creation_timestamp).days
if snapshot_age > max_snapshot_age_days:
report.status = "FAIL"
report.status_extended = (
f"Disk snapshot {snapshot.name} is {snapshot_age} days old, "
f"exceeding the {max_snapshot_age_days} day threshold."
)
else:
report.status = "PASS"
report.status_extended = (
f"Disk snapshot {snapshot.name} is {snapshot_age} days old, "
f"within the {max_snapshot_age_days} day threshold."
)
findings.append(report)
return findings
+6
View File
@@ -64,6 +64,7 @@ def mock_api_client(GCPService, service, api_version, _):
mock_api_access_policies_calls(client)
mock_api_instance_group_managers_calls(client)
mock_api_images_calls(client)
mock_api_snapshots_calls(client)
return client
@@ -1344,3 +1345,8 @@ def mock_api_images_calls(client: MagicMock):
return return_value
client.images().getIamPolicy = mock_get_image_iam_policy
def mock_api_snapshots_calls(client: MagicMock):
client.snapshots().list().execute.return_value = {"items": []}
client.snapshots().list_next.return_value = None
+1
View File
@@ -92,6 +92,7 @@ class TestGCPProvider:
"max_unused_account_days": 180,
"storage_min_retention_days": 90,
"mig_min_zones": 2,
"max_snapshot_age_days": 90,
}
@freeze_time(datetime.today())
@@ -0,0 +1,324 @@
from datetime import datetime, timedelta, timezone
from unittest import mock
from tests.providers.gcp.gcp_fixtures import GCP_PROJECT_ID, set_mocked_gcp_provider
class TestComputeSnapshotNotOutdated:
def test_compute_no_snapshots(self):
compute_client = mock.MagicMock()
compute_client.snapshots = []
compute_client.audit_config = {"max_snapshot_age_days": 90}
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_gcp_provider(),
),
mock.patch(
"prowler.providers.gcp.services.compute.compute_snapshot_not_outdated.compute_snapshot_not_outdated.compute_client",
new=compute_client,
),
):
from prowler.providers.gcp.services.compute.compute_snapshot_not_outdated.compute_snapshot_not_outdated import (
compute_snapshot_not_outdated,
)
check = compute_snapshot_not_outdated()
result = check.execute()
assert len(result) == 0
def test_snapshot_within_threshold(self):
compute_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.compute.compute_snapshot_not_outdated.compute_snapshot_not_outdated.compute_client",
new=compute_client,
),
):
from prowler.providers.gcp.services.compute.compute_service import Snapshot
from prowler.providers.gcp.services.compute.compute_snapshot_not_outdated.compute_snapshot_not_outdated import (
compute_snapshot_not_outdated,
)
creation_time = datetime.now(timezone.utc) - timedelta(days=30)
compute_client.project_ids = [GCP_PROJECT_ID]
compute_client.audit_config = {"max_snapshot_age_days": 90}
compute_client.snapshots = [
Snapshot(
name="test-snapshot-recent",
id="1234567890",
project_id=GCP_PROJECT_ID,
creation_timestamp=creation_time,
source_disk="test-disk",
source_disk_id="disk-123",
disk_size_gb=100,
storage_bytes=1073741824,
storage_locations=["us-central1"],
status="READY",
auto_created=False,
)
]
check = compute_snapshot_not_outdated()
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert "30 days old" in result[0].status_extended
assert "within the 90 day threshold" in result[0].status_extended
assert result[0].resource_id == "1234567890"
assert result[0].resource_name == "test-snapshot-recent"
assert result[0].location == "global"
assert result[0].project_id == GCP_PROJECT_ID
def test_snapshot_exceeds_threshold(self):
compute_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.compute.compute_snapshot_not_outdated.compute_snapshot_not_outdated.compute_client",
new=compute_client,
),
):
from prowler.providers.gcp.services.compute.compute_service import Snapshot
from prowler.providers.gcp.services.compute.compute_snapshot_not_outdated.compute_snapshot_not_outdated import (
compute_snapshot_not_outdated,
)
creation_time = datetime.now(timezone.utc) - timedelta(days=120)
compute_client.project_ids = [GCP_PROJECT_ID]
compute_client.audit_config = {"max_snapshot_age_days": 90}
compute_client.snapshots = [
Snapshot(
name="test-snapshot-old",
id="0987654321",
project_id=GCP_PROJECT_ID,
creation_timestamp=creation_time,
source_disk="test-disk",
source_disk_id="disk-456",
disk_size_gb=200,
storage_bytes=2147483648,
storage_locations=["us-east1"],
status="READY",
auto_created=False,
)
]
check = compute_snapshot_not_outdated()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert "120 days old" in result[0].status_extended
assert "exceeding the 90 day threshold" in result[0].status_extended
assert result[0].resource_id == "0987654321"
assert result[0].resource_name == "test-snapshot-old"
assert result[0].location == "global"
assert result[0].project_id == GCP_PROJECT_ID
def test_snapshot_no_creation_timestamp(self):
compute_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.compute.compute_snapshot_not_outdated.compute_snapshot_not_outdated.compute_client",
new=compute_client,
),
):
from prowler.providers.gcp.services.compute.compute_service import Snapshot
from prowler.providers.gcp.services.compute.compute_snapshot_not_outdated.compute_snapshot_not_outdated import (
compute_snapshot_not_outdated,
)
compute_client.project_ids = [GCP_PROJECT_ID]
compute_client.audit_config = {"max_snapshot_age_days": 90}
compute_client.snapshots = [
Snapshot(
name="test-snapshot-no-timestamp",
id="1111111111",
project_id=GCP_PROJECT_ID,
creation_timestamp=None,
source_disk="test-disk",
source_disk_id="disk-789",
disk_size_gb=50,
storage_bytes=536870912,
storage_locations=["eu-west1"],
status="READY",
auto_created=False,
)
]
check = compute_snapshot_not_outdated()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert "timestamp could not be retrieved" in result[0].status_extended
assert result[0].resource_id == "1111111111"
assert result[0].resource_name == "test-snapshot-no-timestamp"
assert result[0].project_id == GCP_PROJECT_ID
def test_multiple_snapshots_mixed(self):
compute_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.compute.compute_snapshot_not_outdated.compute_snapshot_not_outdated.compute_client",
new=compute_client,
),
):
from prowler.providers.gcp.services.compute.compute_service import Snapshot
from prowler.providers.gcp.services.compute.compute_snapshot_not_outdated.compute_snapshot_not_outdated import (
compute_snapshot_not_outdated,
)
current_time = datetime.now(timezone.utc)
compute_client.project_ids = [GCP_PROJECT_ID]
compute_client.audit_config = {"max_snapshot_age_days": 90}
compute_client.snapshots = [
Snapshot(
name="recent-snapshot",
id="1111111111",
project_id=GCP_PROJECT_ID,
creation_timestamp=current_time - timedelta(days=10),
source_disk="disk-1",
status="READY",
),
Snapshot(
name="old-snapshot",
id="2222222222",
project_id=GCP_PROJECT_ID,
creation_timestamp=current_time - timedelta(days=150),
source_disk="disk-2",
status="READY",
),
Snapshot(
name="boundary-snapshot",
id="3333333333",
project_id=GCP_PROJECT_ID,
creation_timestamp=current_time - timedelta(days=91),
source_disk="disk-3",
status="READY",
),
]
check = compute_snapshot_not_outdated()
result = check.execute()
assert len(result) == 3
recent_result = next(r for r in result if r.resource_id == "1111111111")
old_result = next(r for r in result if r.resource_id == "2222222222")
boundary_result = next(r for r in result if r.resource_id == "3333333333")
assert recent_result.status == "PASS"
assert recent_result.resource_name == "recent-snapshot"
assert old_result.status == "FAIL"
assert old_result.resource_name == "old-snapshot"
assert boundary_result.status == "FAIL"
assert boundary_result.resource_name == "boundary-snapshot"
def test_custom_threshold(self):
compute_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.compute.compute_snapshot_not_outdated.compute_snapshot_not_outdated.compute_client",
new=compute_client,
),
):
from prowler.providers.gcp.services.compute.compute_service import Snapshot
from prowler.providers.gcp.services.compute.compute_snapshot_not_outdated.compute_snapshot_not_outdated import (
compute_snapshot_not_outdated,
)
creation_time = datetime.now(timezone.utc) - timedelta(days=45)
compute_client.project_ids = [GCP_PROJECT_ID]
compute_client.audit_config = {"max_snapshot_age_days": 30}
compute_client.snapshots = [
Snapshot(
name="test-snapshot-custom",
id="4444444444",
project_id=GCP_PROJECT_ID,
creation_timestamp=creation_time,
source_disk="test-disk",
status="READY",
)
]
check = compute_snapshot_not_outdated()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert "45 days old" in result[0].status_extended
assert "exceeding the 30 day threshold" in result[0].status_extended
def test_default_threshold_when_not_configured(self):
compute_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.compute.compute_snapshot_not_outdated.compute_snapshot_not_outdated.compute_client",
new=compute_client,
),
):
from prowler.providers.gcp.services.compute.compute_service import Snapshot
from prowler.providers.gcp.services.compute.compute_snapshot_not_outdated.compute_snapshot_not_outdated import (
compute_snapshot_not_outdated,
)
creation_time = datetime.now(timezone.utc) - timedelta(days=85)
compute_client.project_ids = [GCP_PROJECT_ID]
compute_client.audit_config = {}
compute_client.snapshots = [
Snapshot(
name="test-snapshot-default",
id="5555555555",
project_id=GCP_PROJECT_ID,
creation_timestamp=creation_time,
source_disk="test-disk",
status="READY",
)
]
check = compute_snapshot_not_outdated()
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert "85 days old" in result[0].status_extended
assert "within the 90 day threshold" in result[0].status_extended