feat(gcp): add provider id validation inside test_connection (#5381)

prowler/providers/gcp/exceptions/exceptions.py

@@ -37,6 +37,10 @@ class GCPBaseException(ProwlerException):
             "message": "Error loading static credentials",
             "remediation": "Check the credentials and ensure they are properly set up. client_id, client_secret and refresh_token are required.",
         },
+        (1933, "GCPInvalidAccountCredentials"): {
+            "message": "Provider does not match with the expected project_id",
+            "remediation": "Check the provider and ensure it matches the expected project_id.",
+        },
     }
 
     def __init__(self, code, file=None, original_exception=None, message=None):
@@ -114,3 +118,10 @@ class GCPStaticCredentialsError(GCPCredentialsError):
         super().__init__(
             1932, file=file, original_exception=original_exception, message=message
         )
+
+
+class GCPInvalidAccountCredentials(GCPBaseException):
+    def __init__(self, file=None, original_exception=None, message=None):
+        super().__init__(
+            1933, file=file, original_exception=original_exception, message=message
+        )

prowler/providers/gcp/gcp_provider.py

@@ -1,6 +1,7 @@
 import os
 import re
 import sys
+from typing import Optional
 
 from colorama import Fore, Style
 from google.auth import default, impersonated_credentials, load_credentials_from_dict
@@ -18,6 +19,7 @@ from prowler.providers.gcp.exceptions.exceptions import (
     GCPCloudResourceManagerAPINotUsedError,
     GCPGetProjectError,
     GCPHTTPError,
+    GCPInvalidAccountCredentials,
     GCPLoadCredentialsFromDictError,
     GCPNoAccesibleProjectsError,
     GCPSetUpSessionError,
@@ -83,7 +85,7 @@ class GcpProvider(Provider):
         self._project_ids = []
         self._projects = {}
         self._excluded_project_ids = []
-        accessible_projects = self.get_projects()
+        accessible_projects = self.get_projects(self._session)
         if not accessible_projects:
             logger.critical("No Project IDs can be accessed via Google Credentials.")
             raise GCPNoAccesibleProjectsError(
@@ -296,6 +298,7 @@ class GcpProvider(Provider):
         client_id: str = None,
         client_secret: str = None,
         refresh_token: str = None,
+        provider_id: Optional[str] = None,
     ) -> Connection:
         """
         Test the connection to GCP with the provided credentials file or service account to impersonate.
@@ -305,6 +308,11 @@ class GcpProvider(Provider):
         Args:
             credentials_file: str
             service_account: str
+            raise_on_exception: bool
+            client_id: str
+            client_secret: str
+            refresh_token: str
+            provider_id: Optional[str] -> The provider ID, for GCP it is the project ID
         Returns:
             Connection object with is_connected set to True if the connection is successful, or error set to the exception if the connection fails
         """
@@ -316,9 +324,15 @@ class GcpProvider(Provider):
                     client_id, client_secret, refresh_token
                 )
 
-            session, _ = GcpProvider.setup_session(
+            session, project_id = GcpProvider.setup_session(
                 credentials_file, service_account, gcp_credentials
             )
+            if provider_id and project_id != provider_id:
+                # Logic to check if the provider ID matches the project ID
+                GcpProvider.validate_project_id(
+                    provider_id=provider_id, credentials=session
+                )
+
             service = discovery.build("cloudresourcemanager", "v1", credentials=session)
             request = service.projects().list()
             request.execute()
@@ -351,6 +365,12 @@ class GcpProvider(Provider):
             if raise_on_exception:
                 raise http_error
             return Connection(error=http_error)
+        # Exceptions from validating Provider ID
+        except GCPInvalidAccountCredentials as not_valid_provider_id_error:
+            logger.critical(str(not_valid_provider_id_error))
+            if raise_on_exception:
+                raise not_valid_provider_id_error
+            return Connection(error=not_valid_provider_id_error)
         except Exception as error:
             logger.critical(
                 f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
@@ -383,12 +403,13 @@ class GcpProvider(Provider):
         )
         print_boxes(report_lines, report_title)
 
-    def get_projects(self) -> dict[str, GCPProject]:
+    @staticmethod
+    def get_projects(credentials) -> dict[str, GCPProject]:
         try:
             projects = {}
 
             service = discovery.build(
-                "cloudresourcemanager", "v1", credentials=self.session
+                "cloudresourcemanager", "v1", credentials=credentials
             )
 
             request = service.projects().list()
@@ -528,3 +549,34 @@ class GcpProvider(Provider):
             "refresh_token": refresh_token,
             "type": "authorized_user",
         }
+
+    @staticmethod
+    def validate_project_id(provider_id: str, credentials: str = None) -> None:
+        """
+        Validate the provider ID given the credentials, checking if the provider ID matches with the expected project_id using the method get_projects
+
+        Args:
+            provider_id: str
+            credentials: str
+
+        Returns:
+            None
+
+        Raises:
+            GCPInvalidAccountCredentials if the provider ID does not match with the expected project_id
+        """
+
+        available_projects = list(
+            GcpProvider.get_projects(credentials=credentials).keys()
+        )
+
+        if len(available_projects) == 0:
+            raise GCPNoAccesibleProjectsError(
+                file=__file__,
+                message="No Project IDs can be accessed via Google Credentials.",
+            )
+        elif provider_id not in available_projects:
+            raise GCPInvalidAccountCredentials(
+                file=__file__,
+                message="The provider ID does not match with the expected project_id.",
+            )

tests/providers/gcp/gcp_provider_test.py

@@ -11,7 +11,11 @@ from prowler.config.config import (
     default_fixer_config_file_path,
     load_and_validate_config_file,
 )
-from prowler.providers.gcp.exceptions.exceptions import GCPTestConnectionError
+from prowler.providers.common.models import Connection
+from prowler.providers.gcp.exceptions.exceptions import (
+    GCPInvalidAccountCredentials,
+    GCPTestConnectionError,
+)
 from prowler.providers.gcp.gcp_provider import GcpProvider
 from prowler.providers.gcp.models import GCPIdentityInfo, GCPProject
 
@@ -538,3 +542,67 @@ class TestGCPProvider:
                 )
             assert e.type == GCPTestConnectionError
             assert "Test exception" in e.value.args[0]
+
+    def test_test_connection_valid_project_id(self):
+        project_id = "test-project-id"
+        mocked_service = MagicMock()
+
+        mocked_service.projects.get.return_value = MagicMock(
+            execute=MagicMock(return_value={"projectId": project_id})
+        )
+
+        with patch(
+            "prowler.providers.gcp.gcp_provider.GcpProvider.setup_session",
+            return_value=(None, project_id),
+        ), patch(
+            "prowler.providers.gcp.gcp_provider.discovery.build",
+            return_value=mocked_service,
+        ):
+            output = GcpProvider.test_connection(
+                client_id="test-client-id",
+                client_secret="test-client-secret",
+                refresh_token="test-refresh-token",
+                provider_id=project_id,
+            )
+            assert Connection(is_connected=True, error=None) == output
+
+    def test_test_connection_invalid_project_id(self):
+        mocked_service = MagicMock()
+
+        projects = {
+            "test-valid-project": GCPProject(
+                number="55555555",
+                id="project/55555555",
+                name="test-project",
+                labels={"test": "value"},
+                lifecycle_state="",
+            ),
+        }
+
+        mocked_service.projects.get.return_value = MagicMock(
+            execute=MagicMock(return_value={"projects": projects})
+        )
+
+        with patch(
+            "prowler.providers.gcp.gcp_provider.GcpProvider.setup_session",
+            return_value=(None, "test-valid-project"),
+        ), patch(
+            "prowler.providers.gcp.gcp_provider.discovery.build",
+            return_value=mocked_service,
+        ), patch(
+            "prowler.providers.gcp.gcp_provider.GcpProvider.validate_project_id"
+        ) as mock_validate_project_id:
+
+            mock_validate_project_id.side_effect = GCPInvalidAccountCredentials(
+                "Invalid project ID"
+            )
+
+            with pytest.raises(Exception) as e:
+                GcpProvider.test_connection(
+                    client_id="test-client-id",
+                    client_secret="test-client-secret",
+                    refresh_token="test-refresh-token",
+                    provider_id="test-invalid-project",
+                )
+
+            assert e.type == GCPInvalidAccountCredentials