fix(gcp): surface organization-scan failures instead of silently scanning the home project (#11280)

Co-authored-by: Daniel Barranquero <danielbo2001@gmail.com>

docs/user-guide/providers/gcp/authentication.mdx

@@ -138,6 +138,10 @@ To keep permissions focused:
 
 4. Continue through the wizard and finish. No principals need to be granted access in step 3 unless you want other identities to impersonate this account.
 
+<Note>
+To use this service account with `--organization-id`, additionally grant `roles/cloudasset.viewer` at the organization node and enable the Cloud Asset API in the service account's host project. See [Scanning a Specific GCP Organization](./organization). Without these, organization-wide scans silently fall back to listing only the projects accessible to the service account.
+</Note>
+
 ### Step 3: Generate a JSON Key
 
 1. Open the newly created service account, move to the **Keys** tab, and choose **Add key > Create new key**.

docs/user-guide/providers/gcp/organization.mdx

@@ -11,8 +11,19 @@ prowler gcp --organization-id organization-id
 ```
 
 <Warning>
-Ensure the credentials used have one of the following roles at the organization level:
-Cloud Asset Viewer (`roles/cloudasset.viewer`), or Cloud Asset Owner (`roles/cloudasset.owner`).
+Ensure the credentials used have one of the following roles bound **at the organization node** (not at a project): Cloud Asset Viewer (`roles/cloudasset.viewer`) or Cloud Asset Owner (`roles/cloudasset.owner`). The role must be bound directly on the organization so the Cloud Asset API can enumerate projects across the whole hierarchy.
+
+```bash
+gcloud organizations add-iam-policy-binding <organization-id> \
+  --member="serviceAccount:<service-account-email>" \
+  --role="roles/cloudasset.viewer"
+```
+
+The Cloud Asset API (`cloudasset.googleapis.com`) must also be enabled in the project that owns the credentials (the service account's host project, or the quota project for user credentials):
+
+```bash
+gcloud services enable cloudasset.googleapis.com --project <credentials-project-id>
+```
 
 </Warning>
 <Note>

prowler/CHANGELOG.md

@@ -35,6 +35,7 @@ All notable changes to the **Prowler SDK** are documented in this file.
 ### 🐞 Fixed
 
 - GCP `logging_log_metric_filter_and_alert_*` checks now credit org-level aggregated sinks filtered to the Admin Activity audit stream [(#11575)](https://github.com/prowler-cloud/prowler/pull/11575)
+- GCP organization scans with `--organization-id` no longer silently fall back to the credentials' host project when the Cloud Asset API call fails; the new `GCPGetOrganizationProjectsError` (3011) is raised instead, naming the required `roles/cloudasset.viewer` binding and Cloud Asset API enablement [(#11280)](https://github.com/prowler-cloud/prowler/pull/11280)
 
 ---
 

prowler/providers/gcp/exceptions/exceptions.py

@@ -34,11 +34,17 @@ class GCPBaseException(ProwlerException):
             "message": "Error loading Service Account Private Key credentials from dictionary",
             "remediation": "Check the dictionary and ensure it contains a Service Account Private Key.",
         },
+        (3011, "GCPGetOrganizationProjectsError"): {
+            "message": "Error retrieving projects under the organization via the Cloud Asset API",
+            "remediation": "Ensure the Cloud Asset API is enabled in the credentials' project and that the principal has 'roles/cloudasset.viewer' bound at the organization level. See https://cloud.google.com/asset-inventory/docs/access-control.",
+        },
     }
 
     def __init__(self, code, file=None, original_exception=None, message=None):
         provider = "GCP"
-        error_info = self.GCP_ERROR_CODES.get((code, self.__class__.__name__))
+        # Copy the catalog entry so a custom message does not mutate the
+        # class-level GCP_ERROR_CODES shared across exception instances.
+        error_info = dict(self.GCP_ERROR_CODES.get((code, self.__class__.__name__)))
         if message:
             error_info["message"] = message
         super().__init__(
@@ -104,3 +110,10 @@ class GCPLoadServiceAccountKeyFromDictError(GCPCredentialsError):
         super().__init__(
             3010, file=file, original_exception=original_exception, message=message
         )
+
+
+class GCPGetOrganizationProjectsError(GCPBaseException):
+    def __init__(self, file=None, original_exception=None, message=None):
+        super().__init__(
+            3011, file=file, original_exception=original_exception, message=message
+        )

prowler/providers/gcp/gcp_provider.py

@@ -21,6 +21,8 @@ from prowler.providers.common.models import Audit_Metadata, Connection
 from prowler.providers.common.provider import Provider
 from prowler.providers.gcp.config import DEFAULT_RETRY_ATTEMPTS
 from prowler.providers.gcp.exceptions.exceptions import (
+    GCPBaseException,
+    GCPGetOrganizationProjectsError,
     GCPInvalidProviderIdError,
     GCPLoadADCFromDictError,
     GCPLoadServiceAccountKeyFromDictError,
@@ -621,10 +623,7 @@ class GcpProvider(Provider):
             credentials_file: str
 
         Returns:
-            dict[str, GCPProject]
-
-        Usage:
-            >>> GcpProvider.get_projects(credentials=credentials, organization_id=organization_id)
+            dict of project_id and GCPProject object
         """
         projects = {}
         try:
@@ -632,7 +631,10 @@ class GcpProvider(Provider):
                 try:
                     # Initialize Cloud Asset Inventory API for recursive project retrieval
                     asset_service = discovery.build(
-                        "cloudasset", "v1", credentials=credentials
+                        "cloudasset",
+                        "v1",
+                        credentials=credentials,
+                        num_retries=DEFAULT_RETRY_ATTEMPTS,
                     )
                     # Set the scope to the specified organization and filter for projects
                     scope = f"organizations/{organization_id}"
@@ -643,7 +645,7 @@ class GcpProvider(Provider):
                     )
 
                     while request is not None:
-                        response = request.execute()
+                        response = request.execute(num_retries=DEFAULT_RETRY_ATTEMPTS)
 
                         for asset in response.get("assets", []):
                             # Extract labels and other project details
@@ -688,13 +690,25 @@ class GcpProvider(Provider):
                         )
                 except HttpError as http_error:
                     if "Cloud Asset API has not been used" in str(http_error):
-                        logger.error(
-                            f"Projects cannot be retrieved from the Organization since Cloud Asset API has not been used before or it is disabled [{http_error.__traceback__.tb_lineno}]. Enable it by visiting https://console.developers.google.com/apis/api/cloudasset.googleapis.com/ then retry."
+                        message = (
+                            "Projects cannot be retrieved from the Organization since the Cloud Asset API "
+                            "has not been used before or it is disabled. Enable it by visiting "
+                            "https://console.developers.google.com/apis/api/cloudasset.googleapis.com/ then retry."
                         )
                     else:
-                        logger.error(
-                            f"{http_error.__class__.__name__}[{http_error.__traceback__.tb_lineno}]: {http_error}"
+                        message = (
+                            f"Cloud Asset API call failed while listing projects under organization "
+                            f"'{organization_id}': {http_error}. Ensure the credentials' principal has "
+                            "'roles/cloudasset.viewer' bound at the organization level."
                         )
+                    logger.critical(
+                        f"{http_error.__class__.__name__}[{http_error.__traceback__.tb_lineno}]: {message}"
+                    )
+                    raise GCPGetOrganizationProjectsError(
+                        file=__file__,
+                        original_exception=http_error,
+                        message=message,
+                    )
             else:
                 try:
                     # Initialize Cloud Resource Manager API for simple project listing
@@ -781,8 +795,10 @@ class GcpProvider(Provider):
                             labels={},
                             lifecycle_state="ACTIVE",
                         )
-                # If no projects were able to be accessed via API, add them manually from the credentials file
-                elif credentials_file:
+                # If no projects were able to be accessed via API, add them manually from the credentials file.
+                # Skip this fallback when an organization scan was explicitly requested: silently
+                # downgrading scope to the service account's home project hides permission errors.
+                elif credentials_file and not organization_id:
                     with open(credentials_file, "r", encoding="utf-8") as file:
                         project_id = json.load(file)["project_id"]
                         # Handle empty or null project names
@@ -798,6 +814,8 @@ class GcpProvider(Provider):
                             labels={},
                             lifecycle_state="ACTIVE",
                         )
+        except GCPBaseException as gcp_error:
+            raise gcp_error
         except Exception as error:
             logger.critical(
                 f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"

tests/providers/gcp/gcp_provider_test.py

@@ -13,6 +13,7 @@ from prowler.config.config import (
 )
 from prowler.providers.common.models import Connection
 from prowler.providers.gcp.exceptions.exceptions import (
+    GCPGetOrganizationProjectsError,
     GCPInvalidProviderIdError,
     GCPNoAccesibleProjectsError,
     GCPTestConnectionError,
@@ -1077,3 +1078,66 @@ class TestGCPProvider:
 
             assert gcp_provider.skip_api_check is True
             mocked_is_api_active.assert_not_called()
+
+    def test_get_projects_organization_id_permission_denied_raises(self):
+        """When --organization-id is set and the Cloud Asset API returns a 403,
+        get_projects must raise GCPGetOrganizationProjectsError instead of
+        silently falling back to the service account's home project.
+
+        Regression test for https://github.com/prowler-cloud/prowler/issues/11250.
+        """
+        from googleapiclient.errors import HttpError
+
+        forbidden_response = MagicMock(status=403, reason="Forbidden")
+        http_error = HttpError(
+            resp=forbidden_response,
+            content=b'{"error": {"code": 403, "message": "Permission denied on resource organization"}}',
+            uri="https://cloudasset.googleapis.com/v1/organizations/123:listAssets",
+        )
+
+        asset_service = MagicMock()
+        asset_service.assets.return_value.list.return_value.execute.side_effect = (
+            http_error
+        )
+
+        with patch(
+            "prowler.providers.gcp.gcp_provider.discovery.build",
+            return_value=asset_service,
+        ):
+            with pytest.raises(GCPGetOrganizationProjectsError):
+                GcpProvider.get_projects(
+                    credentials=MagicMock(),
+                    organization_id="test-organization-id",
+                    credentials_file="test_credentials_file",
+                )
+
+    def test_get_projects_organization_id_cloud_asset_api_disabled_raises(self):
+        """When --organization-id is set and the Cloud Asset API is disabled,
+        get_projects must raise GCPGetOrganizationProjectsError with the
+        enable-API remediation rather than swallowing the error."""
+        from googleapiclient.errors import HttpError
+
+        disabled_response = MagicMock(status=403, reason="Forbidden")
+        http_error = HttpError(
+            resp=disabled_response,
+            content=b'{"error": {"message": "Cloud Asset API has not been used in project 123 before or it is disabled."}}',
+            uri="https://cloudasset.googleapis.com/v1/organizations/123:listAssets",
+        )
+
+        asset_service = MagicMock()
+        asset_service.assets.return_value.list.return_value.execute.side_effect = (
+            http_error
+        )
+
+        with patch(
+            "prowler.providers.gcp.gcp_provider.discovery.build",
+            return_value=asset_service,
+        ):
+            with pytest.raises(GCPGetOrganizationProjectsError) as exc_info:
+                GcpProvider.get_projects(
+                    credentials=MagicMock(),
+                    organization_id="test-organization-id",
+                    credentials_file="test_credentials_file",
+                )
+
+        assert "Cloud Asset API" in str(exc_info.value)