From 2cde4c939d7fa67ae1adf642a398109a3f774aa6 Mon Sep 17 00:00:00 2001 From: lydiavilchez <114735608+lydiavilchez@users.noreply.github.com> Date: Wed, 14 Jan 2026 12:35:29 +0100 Subject: [PATCH] feat(gcp): add compute_snapshot_not_outdated check (#9774) --- prowler/CHANGELOG.md | 1 + prowler/config/config.yaml | 3 + .../gcp/services/compute/compute_service.py | 68 ++++ .../compute_snapshot_not_outdated/__init__.py | 0 ...ompute_snapshot_not_outdated.metadata.json | 38 ++ .../compute_snapshot_not_outdated.py | 60 ++++ tests/providers/gcp/gcp_fixtures.py | 6 + tests/providers/gcp/gcp_provider_test.py | 1 + .../compute_snapshot_not_outdated_test.py | 324 ++++++++++++++++++ 9 files changed, 501 insertions(+) create mode 100644 prowler/providers/gcp/services/compute/compute_snapshot_not_outdated/__init__.py create mode 100644 prowler/providers/gcp/services/compute/compute_snapshot_not_outdated/compute_snapshot_not_outdated.metadata.json create mode 100644 prowler/providers/gcp/services/compute/compute_snapshot_not_outdated/compute_snapshot_not_outdated.py create mode 100644 tests/providers/gcp/services/compute/compute_snapshot_not_outdated/compute_snapshot_not_outdated_test.py diff --git a/prowler/CHANGELOG.md b/prowler/CHANGELOG.md index 49112dbd4e..c1db3dc0af 100644 --- a/prowler/CHANGELOG.md +++ b/prowler/CHANGELOG.md @@ -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) diff --git a/prowler/config/config.yaml b/prowler/config/config.yaml index ab5cce8ba9..98636bb0c2 100644 --- a/prowler/config/config.yaml +++ b/prowler/config/config.yaml @@ -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 diff --git a/prowler/providers/gcp/services/compute/compute_service.py b/prowler/providers/gcp/services/compute/compute_service.py index efe40dd765..a1e576f9f9 100644 --- a/prowler/providers/gcp/services/compute/compute_service.py +++ b/prowler/providers/gcp/services/compute/compute_service.py @@ -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 diff --git a/prowler/providers/gcp/services/compute/compute_snapshot_not_outdated/__init__.py b/prowler/providers/gcp/services/compute/compute_snapshot_not_outdated/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/prowler/providers/gcp/services/compute/compute_snapshot_not_outdated/compute_snapshot_not_outdated.metadata.json b/prowler/providers/gcp/services/compute/compute_snapshot_not_outdated/compute_snapshot_not_outdated.metadata.json new file mode 100644 index 0000000000..ddf5920900 --- /dev/null +++ b/prowler/providers/gcp/services/compute/compute_snapshot_not_outdated/compute_snapshot_not_outdated.metadata.json @@ -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." +} diff --git a/prowler/providers/gcp/services/compute/compute_snapshot_not_outdated/compute_snapshot_not_outdated.py b/prowler/providers/gcp/services/compute/compute_snapshot_not_outdated/compute_snapshot_not_outdated.py new file mode 100644 index 0000000000..8537cd390e --- /dev/null +++ b/prowler/providers/gcp/services/compute/compute_snapshot_not_outdated/compute_snapshot_not_outdated.py @@ -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 diff --git a/tests/providers/gcp/gcp_fixtures.py b/tests/providers/gcp/gcp_fixtures.py index 2f968f5555..01e9f033c0 100644 --- a/tests/providers/gcp/gcp_fixtures.py +++ b/tests/providers/gcp/gcp_fixtures.py @@ -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 diff --git a/tests/providers/gcp/gcp_provider_test.py b/tests/providers/gcp/gcp_provider_test.py index e561674d1f..7d2bea9f88 100644 --- a/tests/providers/gcp/gcp_provider_test.py +++ b/tests/providers/gcp/gcp_provider_test.py @@ -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()) diff --git a/tests/providers/gcp/services/compute/compute_snapshot_not_outdated/compute_snapshot_not_outdated_test.py b/tests/providers/gcp/services/compute/compute_snapshot_not_outdated/compute_snapshot_not_outdated_test.py new file mode 100644 index 0000000000..9f0a246f1e --- /dev/null +++ b/tests/providers/gcp/services/compute/compute_snapshot_not_outdated/compute_snapshot_not_outdated_test.py @@ -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