feat(gcp): add check to detect Compute Engine configuration changes

docs/user-guide/cli/tutorials/configuration_file.mdx

@@ -97,6 +97,7 @@ The following list includes all the GCP checks with configurable variables that
 
 | Check Name                                                    | Value                                            | Type            |
 |---------------------------------------------------------------|--------------------------------------------------|-----------------|
+| `compute_configuration_changes`                              | `compute_audit_log_lookback_days`                | Integer         |
 | `compute_instance_group_multiple_zones`                      | `mig_min_zones`                                  | Integer         |
 
 ## Kubernetes
@@ -553,6 +554,9 @@ gcp:
   # GCP Compute Configuration
   # gcp.compute_public_address_shodan
   shodan_api_key: null
+  # gcp.compute_configuration_changes
+  # Number of days to look back for Compute Engine configuration changes in audit logs
+  compute_audit_log_lookback_days: 1
   # gcp.compute_instance_group_multiple_zones
   # Minimum number of zones a MIG should span for high availability
   mig_min_zones: 2

prowler/CHANGELOG.md

@@ -12,6 +12,7 @@ All notable changes to the **Prowler SDK** are documented in this file.
 - `compute_instance_disk_auto_delete_disabled` check for GCP provider [(#9604)](https://github.com/prowler-cloud/prowler/pull/9604)
 - Bedrock service pagination [(#9606)](https://github.com/prowler-cloud/prowler/pull/9606)
 - `ResourceGroup` field to all check metadata for resource classification [(#9656)](https://github.com/prowler-cloud/prowler/pull/9656)
+- `compute_configuration_changes` check for GCP provider to detect Compute Engine configuration changes in Cloud Audit Logs [(#9698)](https://github.com/prowler-cloud/prowler/pull/9698)
 - `compute_instance_group_load_balancer_attached` check for GCP provider [(#9695)](https://github.com/prowler-cloud/prowler/pull/9695)
 
 ### Changed

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_configuration_changes
+  # Number of days to look back for Compute Engine configuration changes in audit logs
+  compute_audit_log_lookback_days: 1
   # GCP Service Account and user-managed keys unused configuration
   # gcp.iam_service_account_unused
   # gcp.iam_sa_user_managed_key_unused

prowler/providers/gcp/services/compute/compute_configuration_changes/compute_configuration_changes.metadata.json

@@ -0,0 +1,39 @@
+{
+  "Provider": "gcp",
+  "CheckID": "compute_configuration_changes",
+  "CheckTitle": "Compute Engine resource has no recent configuration changes in audit logs",
+  "CheckType": [],
+  "ServiceName": "compute",
+  "SubServiceName": "",
+  "ResourceIdTemplate": "",
+  "Severity": "low",
+  "ResourceType": "compute.googleapis.com/Instance",
+  "ResourceGroup": "monitoring",
+  "Description": "This check examines Cloud Audit Logs for recent Compute Engine configuration changes. It surfaces modifications to instance settings, disks, and networks within a configurable lookback window so operators can review unexpected changes.",
+  "Risk": "Unreviewed Compute Engine configuration changes may indicate:\n\n- **Unauthorized access** - Malicious actors modifying resources\n- **Lateral movement** - Attackers expanding their foothold\n- **Security policy violations** - Unapproved changes bypassing change management\n\nWithout monitoring, unexpected changes could compromise security posture.",
+  "RelatedUrl": "",
+  "AdditionalURLs": [
+    "https://www.trendmicro.com/cloudoneconformity/knowledge-base/gcp/ComputeEngine/gcp-compute-engine-configuration-changes.html",
+    "https://cloud.google.com/logging/docs/audit"
+  ],
+  "Remediation": {
+    "Code": {
+      "CLI": "gcloud logging read 'protoPayload.serviceName=\"compute.googleapis.com\" AND logName:\"cloudaudit.googleapis.com%2Factivity\"' --project=PROJECT_ID --limit=100 --format=json",
+      "NativeIaC": "",
+      "Other": "1. Navigate to **Cloud Logging** in the GCP Console\n2. Select **Logs Explorer**\n3. Filter logs with: `protoPayload.serviceName=\"compute.googleapis.com\"`\n4. Review the Admin Activity logs for unexpected changes\n5. Investigate any unauthorized modifications",
+      "Terraform": ""
+    },
+    "Recommendation": {
+      "Text": "Apply the **Principle of Least Privilege** to limit who can modify Compute Engine resources. Configure Cloud Monitoring alerts for configuration changes and establish a formal change management process to review all modifications.",
+      "Url": "https://hub.prowler.com/check/compute_configuration_changes"
+    }
+  },
+  "Categories": [
+    "logging"
+  ],
+  "DependsOn": [],
+  "RelatedTo": [
+    "logging_log_metric_filter_and_alert_for_audit_configuration_changes_enabled"
+  ],
+  "Notes": "This check requires Cloud Audit Logs to be enabled. The lookback window is configurable via the `compute_audit_log_lookback_days` parameter in the configuration file (default: 1 day)."
+}

prowler/providers/gcp/services/compute/compute_configuration_changes/compute_configuration_changes.py

@@ -0,0 +1,59 @@
+from prowler.lib.check.models import Check, Check_Report_GCP
+from prowler.providers.gcp.services.logging.logging_client import logging_client
+
+
+class compute_configuration_changes(Check):
+    """Detect Compute Engine configuration changes in Cloud Audit Logs.
+
+    This check examines Cloud Audit Logs (Admin Activity) for recent Compute Engine
+    configuration changes within a configurable lookback window. It surfaces
+    configuration modifications such as instance settings, disks, and network changes
+    so operators can review unexpected modifications.
+
+    - PASS: No Compute Engine configuration changes detected in the lookback period.
+    - FAIL: Compute Engine configuration changes were detected in the lookback period.
+    """
+
+    def execute(self) -> list[Check_Report_GCP]:
+        findings = []
+
+        for project_id in logging_client.project_ids:
+            audit_entries = logging_client.compute_audit_entries.get(project_id, [])
+
+            if not audit_entries:
+                project_obj = logging_client.projects.get(project_id)
+                report = Check_Report_GCP(
+                    metadata=self.metadata(),
+                    resource=project_obj,
+                    project_id=project_id,
+                    location=logging_client.region,
+                    resource_name=(getattr(project_obj, "name", None) or project_id),
+                    resource_id=project_id,
+                )
+                report.status = "PASS"
+                report.status_extended = f"No Compute Engine configuration changes detected in project {project_id}."
+                findings.append(report)
+            else:
+                for entry in audit_entries:
+                    report = Check_Report_GCP(
+                        metadata=self.metadata(),
+                        resource=entry,
+                        project_id=project_id,
+                        location=logging_client.region,
+                        resource_name=entry.resource_name,
+                        resource_id=entry.insert_id,
+                    )
+                    report.status = "FAIL"
+
+                    actor = entry.principal_email or "unknown actor"
+                    timestamp = entry.timestamp
+                    method = entry.method_name
+
+                    report.status_extended = (
+                        f"Compute Engine configuration change detected: {method} "
+                        f"on resource {entry.resource_name} by {actor} at {timestamp} "
+                        f"in project {project_id}."
+                    )
+                    findings.append(report)
+
+        return findings

prowler/providers/gcp/services/logging/logging_service.py

@@ -1,3 +1,6 @@
+from datetime import datetime, timedelta, timezone
+from typing import Optional
+
 from pydantic.v1 import BaseModel
 
 from prowler.lib.logger import logger
@@ -11,8 +14,10 @@ def __init__(self, provider: GcpProvider):
         super().__init__(__class__.__name__, provider, api_version="v2")
         self.sinks = []
         self.metrics = []
+        self.compute_audit_entries = {}
         self._get_sinks()
         self._get_metrics()
+        self._get_compute_audit_entries()
 
     def _get_sinks(self):
         for project_id in self.project_ids:
@@ -70,6 +75,84 @@ def _get_metrics(self):
                     f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
                 )
 
+    def _get_compute_audit_entries(self):
+        lookback_days = self.audit_config.get("compute_audit_log_lookback_days", 1)
+        start_time = datetime.now(timezone.utc) - timedelta(days=lookback_days)
+        timestamp_filter = start_time.strftime("%Y-%m-%dT%H:%M:%SZ")
+
+        for project_id in self.project_ids:
+            try:
+                self.compute_audit_entries[project_id] = []
+                log_filter = (
+                    f'protoPayload.serviceName="compute.googleapis.com" '
+                    f'AND logName="projects/{project_id}/logs/cloudaudit.googleapis.com%2Factivity" '
+                    f'AND timestamp>="{timestamp_filter}"'
+                )
+
+                request = self.client.entries().list(
+                    body={
+                        "resourceNames": [f"projects/{project_id}"],
+                        "filter": log_filter,
+                        "orderBy": "timestamp desc",
+                        "pageSize": 1000,
+                    }
+                )
+
+                while request is not None:
+                    response = request.execute(num_retries=DEFAULT_RETRY_ATTEMPTS)
+
+                    for entry in response.get("entries", []):
+                        proto_payload = entry.get("protoPayload", {})
+                        resource = entry.get("resource", {})
+                        resource_labels = resource.get("labels", {})
+
+                        auth_info = proto_payload.get("authenticationInfo", {})
+                        request_metadata = proto_payload.get("requestMetadata", {})
+
+                        resource_name = resource_labels.get(
+                            "instance_id",
+                            resource_labels.get(
+                                "disk_id",
+                                resource_labels.get(
+                                    "network_id",
+                                    proto_payload.get("resourceName", "unknown"),
+                                ),
+                            ),
+                        )
+
+                        self.compute_audit_entries[project_id].append(
+                            AuditLogEntry(
+                                insert_id=entry.get("insertId", ""),
+                                timestamp=entry.get("timestamp", ""),
+                                receive_timestamp=entry.get("receiveTimestamp"),
+                                resource_type=resource.get("type", ""),
+                                resource_name=resource_name,
+                                method_name=proto_payload.get("methodName", ""),
+                                service_name=proto_payload.get("serviceName", ""),
+                                principal_email=auth_info.get("principalEmail"),
+                                caller_ip=request_metadata.get("callerIp"),
+                                project_id=project_id,
+                            )
+                        )
+
+                    if "nextPageToken" in response:
+                        request = self.client.entries().list(
+                            body={
+                                "resourceNames": [f"projects/{project_id}"],
+                                "filter": log_filter,
+                                "orderBy": "timestamp desc",
+                                "pageSize": 1000,
+                                "pageToken": response["nextPageToken"],
+                            }
+                        )
+                    else:
+                        request = None
+
+            except Exception as error:
+                logger.error(
+                    f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+                )
+
 
 class Sink(BaseModel):
     name: str
@@ -83,3 +166,16 @@ class Metric(BaseModel):
     type: str
     filter: str
     project_id: str
+
+
+class AuditLogEntry(BaseModel):
+    insert_id: str
+    timestamp: str
+    receive_timestamp: Optional[str] = None
+    resource_type: str
+    resource_name: str
+    method_name: str
+    service_name: str
+    principal_email: Optional[str] = None
+    caller_ip: Optional[str] = None
+    project_id: str

tests/config/config_test.py

@@ -333,6 +333,7 @@ def mock_prowler_get_latest_release(_, **kwargs):
     "shodan_api_key": None,
     "mig_min_zones": 2,
     "max_unused_account_days": 30,
+    "compute_audit_log_lookback_days": 1,
 }
 
 config_kubernetes = {

tests/config/fixtures/config.yaml

@@ -413,6 +413,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_configuration_changes
+  # Number of days to look back for Compute Engine configuration changes in audit logs
+  compute_audit_log_lookback_days: 1
   max_unused_account_days: 30
 
 # Kubernetes Configuration

tests/providers/gcp/gcp_fixtures.py

@@ -32,6 +32,12 @@ def set_mocked_gcp_provider(
     provider.identity = GCPIdentityInfo(
         profile=profile,
     )
+    provider.audit_config = {
+        "compute_audit_log_lookback_days": 1,
+        "mig_min_zones": 2,
+        "max_unused_account_days": 30,
+    }
+    provider.fixer_config = {}
 
     return provider
 
@@ -1102,6 +1108,34 @@ def mock_api_sink_calls(client: MagicMock):
     }
     client.sinks().list_next.return_value = None
 
+    client.entries().list().execute.return_value = {
+        "entries": [
+            {
+                "insertId": "audit-log-entry-1",
+                "timestamp": "2024-01-15T10:30:00Z",
+                "receiveTimestamp": "2024-01-15T10:30:01Z",
+                "resource": {
+                    "type": "gce_instance",
+                    "labels": {
+                        "instance_id": "test-instance-1",
+                        "project_id": GCP_PROJECT_ID,
+                    },
+                },
+                "protoPayload": {
+                    "serviceName": "compute.googleapis.com",
+                    "methodName": "v1.compute.instances.insert",
+                    "resourceName": "projects/test-project/zones/us-central1-a/instances/test-instance-1",
+                    "authenticationInfo": {
+                        "principalEmail": "user@example.com",
+                    },
+                    "requestMetadata": {
+                        "callerIp": "192.168.1.1",
+                    },
+                },
+            },
+        ]
+    }
+
 
 def mock_api_services_calls(client: MagicMock):
     client.services().list().execute.return_value = {

tests/providers/gcp/gcp_provider_test.py

@@ -92,6 +92,7 @@ def test_gcp_provider(self):
                 "max_unused_account_days": 180,
                 "storage_min_retention_days": 90,
                 "mig_min_zones": 2,
+                "compute_audit_log_lookback_days": 1,
             }
 
     @freeze_time(datetime.today())