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

Co-authored-by: Lydia Vilchez <lydiavilchezlopez@gmail.com>
2026-10-09 21:14:22 +00:00 · 2026-06-17 17:35:32 +02:00
parent bae74b8181
commit 5ec4a1cbba
11 changed files with 469 additions and 0 deletions
@@ -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
@@ -0,0 +1,102 @@
+from unittest.mock import MagicMock, patch
+
+from prowler.providers.gcp.services.cloudfunction.cloudfunction_service import (
+    CloudFunction,
+)
+from tests.providers.gcp.gcp_fixtures import (
+    GCP_PROJECT_ID,
+    mock_is_api_active,
+    set_mocked_gcp_provider,
+)
+
+_LOCATION_ID = "us-central1"
+_FUNCTION_NAME = "my-function"
+_CONNECTOR = (
+    f"projects/{GCP_PROJECT_ID}/locations/{_LOCATION_ID}/connectors/my-connector"
+)
+
+
+def _make_cloudfunction_client(functions_list):
+    """Return a mock GCP API client for the Cloud Functions v2 service."""
+    client = MagicMock()
+
+    client.projects().locations().list().execute.return_value = {
+        "locations": [{"locationId": _LOCATION_ID}]
+    }
+    client.projects().locations().list_next.return_value = None
+
+    client.projects().locations().functions().list().execute.return_value = {
+        "functions": functions_list
+    }
+    client.projects().locations().functions().list_next.return_value = None
+
+    return client
+
+
+class TestCloudFunctionService:
+    def test_get_functions_with_vpc_connector(self):
+        def mock_api_client(*args, **kwargs):
+            return _make_cloudfunction_client(
+                functions_list=[
+                    {
+                        "name": f"projects/{GCP_PROJECT_ID}/locations/{_LOCATION_ID}/functions/{_FUNCTION_NAME}",
+                        "state": "ACTIVE",
+                        "serviceConfig": {
+                            "vpcConnector": _CONNECTOR,
+                        },
+                    }
+                ]
+            )
+
+        with (
+            patch(
+                "prowler.providers.gcp.lib.service.service.GCPService.__is_api_active__",
+                new=mock_is_api_active,
+            ),
+            patch(
+                "prowler.providers.gcp.lib.service.service.GCPService.__generate_client__",
+                new=mock_api_client,
+            ),
+        ):
+            cf_client = CloudFunction(
+                set_mocked_gcp_provider(project_ids=[GCP_PROJECT_ID])
+            )
+
+            assert len(cf_client.functions) == 1
+            fn = cf_client.functions[0]
+            assert fn.name == _FUNCTION_NAME
+            assert fn.project_id == GCP_PROJECT_ID
+            assert fn.location == _LOCATION_ID
+            assert fn.state == "ACTIVE"
+            assert fn.vpc_connector == _CONNECTOR
+
+    def test_get_functions_without_vpc_connector(self):
+        def mock_api_client(*args, **kwargs):
+            return _make_cloudfunction_client(
+                functions_list=[
+                    {
+                        "name": f"projects/{GCP_PROJECT_ID}/locations/{_LOCATION_ID}/functions/no-vpc-func",
+                        "state": "ACTIVE",
+                        "serviceConfig": {},
+                    }
+                ]
+            )
+
+        with (
+            patch(
+                "prowler.providers.gcp.lib.service.service.GCPService.__is_api_active__",
+                new=mock_is_api_active,
+            ),
+            patch(
+                "prowler.providers.gcp.lib.service.service.GCPService.__generate_client__",
+                new=mock_api_client,
+            ),
+        ):
+            cf_client = CloudFunction(
+                set_mocked_gcp_provider(project_ids=[GCP_PROJECT_ID])
+            )
+
+            assert len(cf_client.functions) == 1
+            fn = cf_client.functions[0]
+            assert fn.name == "no-vpc-func"
+            assert fn.vpc_connector is None