feat(gcp): add cloudfunction_function_inside_vpc check (#11021)

Co-authored-by: Lydia Vilchez <lydiavilchezlopez@gmail.com>

Author: s1ns3nz0

prowler/CHANGELOG.md

@@ -11,6 +11,7 @@ All notable changes to the **Prowler SDK** are documented in this file.
 - `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)
+- `cloudfunction_function_inside_vpc` check for GCP provider, verifying Cloud Functions have a Serverless VPC Access connector for private egress [(#11021)](https://github.com/prowler-cloud/prowler/pull/11021)
 - `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)
 - `cosmosdb_account_automatic_failover_enabled` check for Azure provider [(#11031)](https://github.com/prowler-cloud/prowler/pull/11031)
 - `cosmosdb_account_backup_policy_continuous` check for Azure provider [(#11032)](https://github.com/prowler-cloud/prowler/pull/11032)

prowler/providers/gcp/services/cloudfunction/__init__.py

@@ -0,0 +1,6 @@
+from prowler.providers.common.provider import Provider
+from prowler.providers.gcp.services.cloudfunction.cloudfunction_service import (
+    CloudFunction,
+)
+
+cloudfunction_client = CloudFunction(Provider.get_global_provider())

prowler/providers/gcp/services/cloudfunction/cloudfunction_client.py

@@ -0,0 +1,40 @@
+{
+  "Provider": "gcp",
+  "CheckID": "cloudfunction_function_inside_vpc",
+  "CheckTitle": "Cloud Function is connected to a VPC network",
+  "CheckType": [],
+  "ServiceName": "cloudfunction",
+  "SubServiceName": "",
+  "ResourceIdTemplate": "",
+  "Severity": "medium",
+  "ResourceType": "cloudfunctions.googleapis.com/Function",
+  "Description": "Cloud Functions are attached to a **Serverless VPC Access connector** so egress traffic is routed through a private VPC network instead of the public internet.\n\nThe evaluation reviews each function's network configuration to confirm that a connector is configured.",
+  "Risk": "Without a VPC connector, Cloud Functions cannot privately reach internal resources such as `Cloud SQL`, `Memorystore`, or `GKE`, forcing those services to be exposed over public IPs. This expands the **attack surface**, weakens **confidentiality** of internal traffic, and breaks network segmentation controls required by most security frameworks.",
+  "RelatedUrl": "",
+  "AdditionalURLs": [
+    "https://cloud.google.com/functions/docs/networking/connecting-vpc",
+    "https://cloud.google.com/vpc/docs/serverless-vpc-access"
+  ],
+  "Remediation": {
+    "Code": {
+      "CLI": "gcloud functions deploy <FUNCTION_NAME> --region=<REGION> --vpc-connector=projects/<PROJECT_ID>/locations/<REGION>/connectors/<CONNECTOR_NAME> --egress-settings=all-traffic",
+      "NativeIaC": "",
+      "Other": "1. In Google Cloud Console, go to Cloud Functions\n2. Select the function and click Edit\n3. Under Connections, select the VPC connector for your network\n4. Set Egress settings to route all traffic through the VPC connector\n5. Save and redeploy the function",
+      "Terraform": "```hcl\nresource \"google_cloudfunctions2_function\" \"<example_resource_name>\" {\n  name     = \"<example_resource_name>\"\n  location = \"us-central1\"\n\n  service_config {\n    vpc_connector                 = \"<example_resource_id>\" # Critical: routes egress through the VPC\n    vpc_connector_egress_settings = \"ALL_TRAFFIC\"\n  }\n}\n```"
+    },
+    "Recommendation": {
+      "Text": "Apply **defense in depth** by routing Cloud Function egress through a **Serverless VPC Access connector** when the function must reach internal resources.\n\nScope each connector to **least privilege** subnets so functions cannot reach unintended endpoints.",
+      "Url": "https://hub.prowler.com/check/cloudfunction_function_inside_vpc"
+    }
+  },
+  "Categories": [
+    "trust-boundaries"
+  ],
+  "DependsOn": [],
+  "RelatedTo": [
+    "cloudfunction_function_not_publicly_accessible",
+    "cloudsql_instance_public_ip",
+    "compute_instance_public_ip"
+  ],
+  "Notes": "A VPC connector must be created in the same region as the Cloud Function. This check only verifies that a connector is attached; it does not validate egress settings or connector configuration."
+}

prowler/providers/gcp/services/cloudfunction/cloudfunction_function_inside_vpc/__init__.py

@@ -0,0 +1,39 @@
+from prowler.lib.check.models import Check, Check_Report_GCP
+from prowler.providers.gcp.services.cloudfunction.cloudfunction_client import (
+    cloudfunction_client,
+)
+
+
+class cloudfunction_function_inside_vpc(Check):
+    """Check that Cloud Functions are attached to a Serverless VPC Access connector.
+
+    Verifies that each active Cloud Function has a `vpcConnector` configured so
+    egress traffic flows through a private VPC network instead of the public
+    internet. Functions in non-`ACTIVE` states are skipped because their network
+    configuration is transient.
+    """
+
+    def execute(self) -> list[Check_Report_GCP]:
+        """Execute the VPC-connector check across all Cloud Functions.
+
+        Returns:
+            A list of `Check_Report_GCP` findings, one per active Cloud
+            Function. Status is `PASS` when a `vpc_connector` is set and `FAIL`
+            otherwise.
+        """
+        findings = []
+        for function in cloudfunction_client.functions:
+            if function.state != "ACTIVE":
+                continue
+            report = Check_Report_GCP(metadata=self.metadata(), resource=function)
+            if function.vpc_connector:
+                report.status = "PASS"
+                report.status_extended = (
+                    f"Cloud Function {function.name} is connected to a VPC via "
+                    f"connector: {function.vpc_connector}."
+                )
+            else:
+                report.status = "FAIL"
+                report.status_extended = f"Cloud Function {function.name} is not connected to any VPC network."
+            findings.append(report)
+        return findings

prowler/providers/gcp/services/cloudfunction/cloudfunction_function_inside_vpc/cloudfunction_function_inside_vpc.metadata.json

@@ -0,0 +1,84 @@
+from typing import Optional
+
+from pydantic.v1 import BaseModel
+
+from prowler.lib.logger import logger
+from prowler.providers.gcp.config import DEFAULT_RETRY_ATTEMPTS
+from prowler.providers.gcp.gcp_provider import GcpProvider
+from prowler.providers.gcp.lib.service.service import GCPService
+
+
+class CloudFunction(GCPService):
+    """Cloud Functions v2 service client.
+
+    Enumerates Cloud Functions across every accessible project and region
+    using the `cloudfunctions.googleapis.com` v2 API and exposes them through
+    the `functions` attribute.
+    """
+
+    def __init__(self, provider: GcpProvider) -> None:
+        """Initialize the service and preload Cloud Functions."""
+        super().__init__("cloudfunctions", provider, api_version="v2")
+        self.functions = []
+        self._get_functions()
+
+    def _get_functions(self) -> None:
+        """Fetch Cloud Functions for every project and location."""
+        for project_id in self.project_ids:
+            try:
+                locations = self.client.projects().locations()
+                locations_request = locations.list(name=f"projects/{project_id}")
+                while locations_request is not None:
+                    locations_response = locations_request.execute(
+                        num_retries=DEFAULT_RETRY_ATTEMPTS
+                    )
+                    for location in locations_response.get("locations", []):
+                        location_id = location["locationId"]
+                        try:
+                            functions = locations.functions()
+                            request = functions.list(
+                                parent=f"projects/{project_id}/locations/{location_id}"
+                            )
+                            while request is not None:
+                                response = request.execute(
+                                    num_retries=DEFAULT_RETRY_ATTEMPTS
+                                )
+                                for fn in response.get("functions", []):
+                                    service_config = fn.get("serviceConfig", {})
+                                    self.functions.append(
+                                        Function(
+                                            name=fn["name"].split("/")[-1],
+                                            project_id=project_id,
+                                            location=location_id,
+                                            state=fn.get("state", "UNKNOWN"),
+                                            vpc_connector=service_config.get(
+                                                "vpcConnector"
+                                            ),
+                                        )
+                                    )
+                                request = functions.list_next(
+                                    previous_request=request,
+                                    previous_response=response,
+                                )
+                        except Exception as error:
+                            logger.error(
+                                f"{location_id} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+                            )
+                    locations_request = locations.list_next(
+                        previous_request=locations_request,
+                        previous_response=locations_response,
+                    )
+            except Exception as error:
+                logger.error(
+                    f"{project_id} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+                )
+
+
+class Function(BaseModel):
+    """Cloud Function resource consumed by GCP checks."""
+
+    name: str
+    project_id: str
+    location: str
+    state: str
+    vpc_connector: Optional[str] = None

prowler/providers/gcp/services/cloudfunction/cloudfunction_function_inside_vpc/cloudfunction_function_inside_vpc.py

@@ -0,0 +1,197 @@
+from unittest import mock
+
+from tests.providers.gcp.gcp_fixtures import (
+    GCP_PROJECT_ID,
+    GCP_US_CENTER1_LOCATION,
+    set_mocked_gcp_provider,
+)
+
+_CHECK_PATH = (
+    "prowler.providers.gcp.services.cloudfunction."
+    "cloudfunction_function_inside_vpc.cloudfunction_function_inside_vpc"
+)
+_CLIENT_PATH = f"{_CHECK_PATH}.cloudfunction_client"
+
+
+class Test_cloudfunction_function_inside_vpc:
+    def test_no_functions(self):
+        cloudfunction_client = mock.MagicMock()
+
+        with (
+            mock.patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=set_mocked_gcp_provider(),
+            ),
+            mock.patch(
+                _CLIENT_PATH,
+                new=cloudfunction_client,
+            ),
+        ):
+            from prowler.providers.gcp.services.cloudfunction.cloudfunction_function_inside_vpc.cloudfunction_function_inside_vpc import (
+                cloudfunction_function_inside_vpc,
+            )
+
+            cloudfunction_client.functions = []
+
+            check = cloudfunction_function_inside_vpc()
+            result = check.execute()
+            assert len(result) == 0
+
+    def test_function_with_vpc_connector(self):
+        cloudfunction_client = mock.MagicMock()
+
+        with (
+            mock.patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=set_mocked_gcp_provider(),
+            ),
+            mock.patch(
+                _CLIENT_PATH,
+                new=cloudfunction_client,
+            ),
+        ):
+            from prowler.providers.gcp.services.cloudfunction.cloudfunction_function_inside_vpc.cloudfunction_function_inside_vpc import (
+                cloudfunction_function_inside_vpc,
+            )
+            from prowler.providers.gcp.services.cloudfunction.cloudfunction_service import (
+                Function,
+            )
+
+            connector = (
+                f"projects/{GCP_PROJECT_ID}/locations/{GCP_US_CENTER1_LOCATION}"
+                f"/connectors/my-connector"
+            )
+            cloudfunction_client.functions = [
+                Function(
+                    name="fn-vpc",
+                    project_id=GCP_PROJECT_ID,
+                    location=GCP_US_CENTER1_LOCATION,
+                    state="ACTIVE",
+                    vpc_connector=connector,
+                )
+            ]
+
+            check = cloudfunction_function_inside_vpc()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "PASS"
+            assert (
+                result[0].status_extended
+                == f"Cloud Function fn-vpc is connected to a VPC via connector: {connector}."
+            )
+            assert result[0].resource_id == "fn-vpc"
+            assert result[0].resource_name == "fn-vpc"
+            assert result[0].location == GCP_US_CENTER1_LOCATION
+            assert result[0].project_id == GCP_PROJECT_ID
+
+    def test_function_without_vpc_connector(self):
+        cloudfunction_client = mock.MagicMock()
+
+        with (
+            mock.patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=set_mocked_gcp_provider(),
+            ),
+            mock.patch(
+                _CLIENT_PATH,
+                new=cloudfunction_client,
+            ),
+        ):
+            from prowler.providers.gcp.services.cloudfunction.cloudfunction_function_inside_vpc.cloudfunction_function_inside_vpc import (
+                cloudfunction_function_inside_vpc,
+            )
+            from prowler.providers.gcp.services.cloudfunction.cloudfunction_service import (
+                Function,
+            )
+
+            cloudfunction_client.functions = [
+                Function(
+                    name="fn-public",
+                    project_id=GCP_PROJECT_ID,
+                    location=GCP_US_CENTER1_LOCATION,
+                    state="ACTIVE",
+                    vpc_connector=None,
+                )
+            ]
+
+            check = cloudfunction_function_inside_vpc()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert (
+                result[0].status_extended
+                == "Cloud Function fn-public is not connected to any VPC network."
+            )
+            assert result[0].resource_id == "fn-public"
+            assert result[0].resource_name == "fn-public"
+            assert result[0].location == GCP_US_CENTER1_LOCATION
+            assert result[0].project_id == GCP_PROJECT_ID
+
+    def test_function_with_empty_vpc_connector(self):
+        cloudfunction_client = mock.MagicMock()
+
+        with (
+            mock.patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=set_mocked_gcp_provider(),
+            ),
+            mock.patch(
+                _CLIENT_PATH,
+                new=cloudfunction_client,
+            ),
+        ):
+            from prowler.providers.gcp.services.cloudfunction.cloudfunction_function_inside_vpc.cloudfunction_function_inside_vpc import (
+                cloudfunction_function_inside_vpc,
+            )
+            from prowler.providers.gcp.services.cloudfunction.cloudfunction_service import (
+                Function,
+            )
+
+            cloudfunction_client.functions = [
+                Function(
+                    name="fn-empty",
+                    project_id=GCP_PROJECT_ID,
+                    location=GCP_US_CENTER1_LOCATION,
+                    state="ACTIVE",
+                    vpc_connector="",
+                )
+            ]
+
+            check = cloudfunction_function_inside_vpc()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+
+    def test_inactive_function_skipped(self):
+        cloudfunction_client = mock.MagicMock()
+
+        with (
+            mock.patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=set_mocked_gcp_provider(),
+            ),
+            mock.patch(
+                _CLIENT_PATH,
+                new=cloudfunction_client,
+            ),
+        ):
+            from prowler.providers.gcp.services.cloudfunction.cloudfunction_function_inside_vpc.cloudfunction_function_inside_vpc import (
+                cloudfunction_function_inside_vpc,
+            )
+            from prowler.providers.gcp.services.cloudfunction.cloudfunction_service import (
+                Function,
+            )
+
+            cloudfunction_client.functions = [
+                Function(
+                    name="fn-deploy",
+                    project_id=GCP_PROJECT_ID,
+                    location=GCP_US_CENTER1_LOCATION,
+                    state="DEPLOYING",
+                    vpc_connector=None,
+                )
+            ]
+
+            check = cloudfunction_function_inside_vpc()
+            result = check.execute()
+            assert len(result) == 0