fix(gcp): honour org-level aggregated sinks in logging_sink_created check (#11355)

Signed-off-by: Oleksandr Sanin <alexaaander.sanin@gmail.com>
Co-authored-by: Hugo P.Brito <hugopbrit@gmail.com>

prowler/CHANGELOG.md

@@ -11,6 +11,14 @@ All notable changes to the **Prowler SDK** are documented in this file.
 
 ---
 
+## [5.29.3] (Prowler UNRELEASED)
+
+### 🐞 Fixed
+
+- GCP `logging_sink_created` now recognizes organization-level aggregated sinks with `includeChildren=True`, avoiding false failures for covered projects [(#11355)](https://github.com/prowler-cloud/prowler/pull/11355)
+
+---
+
 ## [5.29.1] (Prowler v5.29.1)
 
 ### 🐞 Fixed

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

@@ -12,6 +12,7 @@ class Logging(GCPService):
         self.sinks = []
         self.metrics = []
         self._get_sinks()
+        self._get_org_sinks()
         self._get_metrics()
 
     def _get_sinks(self):
@@ -39,6 +40,38 @@ class Logging(GCPService):
                     f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
                 )
 
+    def _get_org_sinks(self):
+        """Fetch org-level sinks with includeChildren so child projects are not falsely failed."""
+        org_ids = set()
+        for project in self.projects.values():
+            if project.organization:
+                org_ids.add(project.organization.id)
+
+        for org_id in org_ids:
+            try:
+                request = self.client.sinks().list(parent=f"organizations/{org_id}")
+                while request is not None:
+                    response = request.execute(num_retries=DEFAULT_RETRY_ATTEMPTS)
+
+                    for sink in response.get("sinks", []):
+                        self.sinks.append(
+                            Sink(
+                                name=sink["name"],
+                                destination=sink["destination"],
+                                filter=sink.get("filter", "all"),
+                                project_id=f"organizations/{org_id}",
+                                include_children=sink.get("includeChildren", False),
+                            )
+                        )
+
+                    request = self.client.sinks().list_next(
+                        previous_request=request, previous_response=response
+                    )
+            except Exception as error:
+                logger.error(
+                    f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+                )
+
     def _get_metrics(self):
         for project_id in self.project_ids:
             try:
@@ -76,6 +109,7 @@ class Sink(BaseModel):
     destination: str
     filter: str
     project_id: str
+    include_children: bool = False
 
 
 class Metric(BaseModel):

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

@@ -5,26 +5,30 @@ from prowler.providers.gcp.services.logging.logging_client import logging_client
 class logging_sink_created(Check):
     def execute(self) -> Check_Report_GCP:
         findings = []
+
+        # Map project_id -> sink for direct project-level sinks
         projects_with_logging_sink = {}
         for sink in logging_client.sinks:
-            if sink.filter == "all":
+            if sink.filter == "all" and not sink.include_children:
                 projects_with_logging_sink[sink.project_id] = sink
 
+        # Collect org resource names that have a covering sink (includeChildren=True)
+        covering_org_sinks = {}
+        for sink in logging_client.sinks:
+            if sink.filter == "all" and sink.include_children:
+                covering_org_sinks[sink.project_id] = sink
+
         for project in logging_client.project_ids:
-            if project not in projects_with_logging_sink.keys():
-                project_obj = logging_client.projects.get(project)
-                report = Check_Report_GCP(
-                    metadata=self.metadata(),
-                    resource=project_obj,
-                    resource_id=project,
-                    project_id=project,
-                    location=logging_client.region,
-                    resource_name=(getattr(project_obj, "name", None) or "GCP Project"),
-                )
-                report.status = "FAIL"
-                report.status_extended = f"There are no logging sinks to export copies of all the log entries in project {project}."
-                findings.append(report)
-            else:
+            project_obj = logging_client.projects.get(project)
+
+            # Determine whether this project is covered by an org-level sink
+            org = getattr(project_obj, "organization", None) if project_obj else None
+            org_resource = f"organizations/{org.id}" if org else None
+            covering_sink = (
+                covering_org_sinks.get(org_resource) if org_resource else None
+            )
+
+            if project in projects_with_logging_sink:
                 sink = projects_with_logging_sink[project]
                 sink_name = getattr(sink, "name", None) or "unknown"
                 report = Check_Report_GCP(
@@ -40,4 +44,31 @@ class logging_sink_created(Check):
                 report.status = "PASS"
                 report.status_extended = f"Sink {sink_name} is enabled exporting copies of all the log entries in project {project}."
                 findings.append(report)
+            elif covering_sink:
+                sink_name = getattr(covering_sink, "name", None) or "unknown"
+                report = Check_Report_GCP(
+                    metadata=self.metadata(),
+                    resource=covering_sink,
+                    resource_id=sink_name,
+                    project_id=project,
+                    location=logging_client.region,
+                    resource_name=(
+                        sink_name if sink_name != "unknown" else "Logging Sink"
+                    ),
+                )
+                report.status = "PASS"
+                report.status_extended = f"Sink {sink_name} at organization level is exporting copies of all the log entries in project {project}."
+                findings.append(report)
+            else:
+                report = Check_Report_GCP(
+                    metadata=self.metadata(),
+                    resource=project_obj,
+                    resource_id=project,
+                    project_id=project,
+                    location=logging_client.region,
+                    resource_name=(getattr(project_obj, "name", None) or "GCP Project"),
+                )
+                report.status = "FAIL"
+                report.status_extended = f"There are no logging sinks to export copies of all the log entries in project {project}."
+                findings.append(report)
         return findings

tests/providers/gcp/services/logging/logging_service_test.py

@@ -1,4 +1,4 @@
-from unittest.mock import patch
+from unittest.mock import MagicMock, patch
 
 from prowler.providers.gcp.services.logging.logging_service import Logging
 from tests.providers.gcp.gcp_fixtures import (
@@ -66,3 +66,74 @@ class TestLoggingService:
                 == "resource.type=gae_app AND severity>=ERROR"
             )
             assert logging_client.metrics[1].project_id == GCP_PROJECT_ID
+
+    def test_org_sinks_fetched_when_project_has_organization(self):
+        """_get_org_sinks() appends org-level sinks when projects have an org."""
+        from prowler.providers.gcp.models import GCPOrganization, GCPProject
+
+        org_id = "999888777"
+        provider = set_mocked_gcp_provider(project_ids=[GCP_PROJECT_ID])
+        provider.projects = {
+            GCP_PROJECT_ID: GCPProject(
+                id=GCP_PROJECT_ID,
+                number="123456789012",
+                name="test",
+                labels={},
+                lifecycle_state="ACTIVE",
+                organization=GCPOrganization(id=org_id, name=f"organizations/{org_id}"),
+            )
+        }
+
+        mock_client = MagicMock()
+        mock_client.sinks().list().execute.return_value = {
+            "sinks": [
+                {
+                    "name": "org-sink",
+                    "destination": "storage.googleapis.com/org-bucket",
+                    "filter": "all",
+                    "includeChildren": True,
+                }
+            ]
+        }
+        mock_client.sinks().list_next.return_value = None
+        mock_client.projects().metrics().list().execute.return_value = {"metrics": []}
+        mock_client.projects().metrics().list_next.return_value = None
+
+        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__",
+                return_value=mock_client,
+            ),
+        ):
+            logging_svc = Logging(provider)
+
+        org_sinks = [
+            s for s in logging_svc.sinks if s.project_id == f"organizations/{org_id}"
+        ]
+        assert len(org_sinks) == 1
+        assert org_sinks[0].name == "org-sink"
+        assert org_sinks[0].include_children is True
+        assert org_sinks[0].filter == "all"
+
+    def test_org_sinks_skipped_when_no_organization(self):
+        """_get_org_sinks() adds nothing when projects have no organization."""
+        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,
+            ),
+        ):
+            logging_svc = Logging(set_mocked_gcp_provider(project_ids=[GCP_PROJECT_ID]))
+
+        org_sinks = [
+            s for s in logging_svc.sinks if s.project_id.startswith("organizations/")
+        ]
+        assert org_sinks == []

tests/providers/gcp/services/logging/logging_sink_created/logging_sink_created_test.py

@@ -1,6 +1,6 @@
 from unittest.mock import MagicMock, patch
 
-from prowler.providers.gcp.models import GCPProject
+from prowler.providers.gcp.models import GCPOrganization, GCPProject
 from tests.providers.gcp.gcp_fixtures import (
     GCP_EU1_LOCATION,
     GCP_PROJECT_ID,
@@ -268,6 +268,7 @@ class Test_logging_sink_created:
             sink.name = None
             sink.filter = "all"
             sink.project_id = GCP_PROJECT_ID
+            sink.include_children = False
 
             logging_client.project_ids = [GCP_PROJECT_ID]
             logging_client.region = GCP_EU1_LOCATION
@@ -311,9 +312,10 @@ class Test_logging_sink_created:
             )
 
             # Create a MagicMock sink object without name attribute
-            sink = MagicMock(spec=["filter", "project_id"])
+            sink = MagicMock(spec=["filter", "project_id", "include_children"])
             sink.filter = "all"
             sink.project_id = GCP_PROJECT_ID
+            sink.include_children = False
 
             logging_client.project_ids = [GCP_PROJECT_ID]
             logging_client.region = GCP_EU1_LOCATION
@@ -336,3 +338,175 @@ class Test_logging_sink_created:
             assert result[0].resource_id == "unknown"
             assert result[0].project_id == GCP_PROJECT_ID
             assert result[0].location == GCP_EU1_LOCATION
+
+    def test_org_level_sink_with_include_children_passes(self):
+        """Projects covered by an org-level sink with includeChildren=True should PASS."""
+        logging_client = MagicMock()
+        org_id = "111222333"
+
+        with (
+            patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=set_mocked_gcp_provider(),
+            ),
+            patch(
+                "prowler.providers.gcp.services.logging.logging_sink_created.logging_sink_created.logging_client",
+                new=logging_client,
+            ),
+        ):
+            from prowler.providers.gcp.services.logging.logging_service import Sink
+            from prowler.providers.gcp.services.logging.logging_sink_created.logging_sink_created import (
+                logging_sink_created,
+            )
+
+            logging_client.project_ids = [GCP_PROJECT_ID]
+            logging_client.region = GCP_EU1_LOCATION
+            logging_client.sinks = [
+                Sink(
+                    name="org-sink",
+                    destination="storage.googleapis.com/org-bucket",
+                    filter="all",
+                    project_id=f"organizations/{org_id}",
+                    include_children=True,
+                )
+            ]
+            logging_client.projects = {
+                GCP_PROJECT_ID: GCPProject(
+                    id=GCP_PROJECT_ID,
+                    number="123456789012",
+                    name="test",
+                    labels={},
+                    lifecycle_state="ACTIVE",
+                    organization=GCPOrganization(
+                        id=org_id, name=f"organizations/{org_id}"
+                    ),
+                )
+            }
+
+            check = logging_sink_created()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "PASS"
+            assert (
+                result[0].status_extended
+                == f"Sink org-sink at organization level is exporting copies of all the log entries in project {GCP_PROJECT_ID}."
+            )
+            assert result[0].resource_id == "org-sink"
+            assert result[0].project_id == GCP_PROJECT_ID
+            assert result[0].location == GCP_EU1_LOCATION
+
+    def test_org_level_sink_without_include_children_fails(self):
+        """Projects NOT covered by includeChildren should still FAIL if no direct project sink."""
+        logging_client = MagicMock()
+        org_id = "111222333"
+
+        with (
+            patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=set_mocked_gcp_provider(),
+            ),
+            patch(
+                "prowler.providers.gcp.services.logging.logging_sink_created.logging_sink_created.logging_client",
+                new=logging_client,
+            ),
+        ):
+            from prowler.providers.gcp.services.logging.logging_service import Sink
+            from prowler.providers.gcp.services.logging.logging_sink_created.logging_sink_created import (
+                logging_sink_created,
+            )
+
+            logging_client.project_ids = [GCP_PROJECT_ID]
+            logging_client.region = GCP_EU1_LOCATION
+            logging_client.sinks = [
+                Sink(
+                    name="org-sink-no-children",
+                    destination="storage.googleapis.com/org-bucket",
+                    filter="all",
+                    project_id=f"organizations/{org_id}",
+                    include_children=False,
+                )
+            ]
+            logging_client.projects = {
+                GCP_PROJECT_ID: GCPProject(
+                    id=GCP_PROJECT_ID,
+                    number="123456789012",
+                    name="test",
+                    labels={},
+                    lifecycle_state="ACTIVE",
+                    organization=GCPOrganization(
+                        id=org_id, name=f"organizations/{org_id}"
+                    ),
+                )
+            }
+
+            check = logging_sink_created()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert (
+                result[0].status_extended
+                == f"There are no logging sinks to export copies of all the log entries in project {GCP_PROJECT_ID}."
+            )
+            assert result[0].resource_id == GCP_PROJECT_ID
+            assert result[0].project_id == GCP_PROJECT_ID
+
+    def test_project_sink_takes_precedence_over_org_sink(self):
+        """A direct project sink should be reported even when an org-level sink also covers the project."""
+        logging_client = MagicMock()
+        org_id = "111222333"
+
+        with (
+            patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=set_mocked_gcp_provider(),
+            ),
+            patch(
+                "prowler.providers.gcp.services.logging.logging_sink_created.logging_sink_created.logging_client",
+                new=logging_client,
+            ),
+        ):
+            from prowler.providers.gcp.services.logging.logging_service import Sink
+            from prowler.providers.gcp.services.logging.logging_sink_created.logging_sink_created import (
+                logging_sink_created,
+            )
+
+            logging_client.project_ids = [GCP_PROJECT_ID]
+            logging_client.region = GCP_EU1_LOCATION
+            logging_client.sinks = [
+                Sink(
+                    name="project-sink",
+                    destination="storage.googleapis.com/project-bucket",
+                    filter="all",
+                    project_id=GCP_PROJECT_ID,
+                ),
+                Sink(
+                    name="org-sink",
+                    destination="storage.googleapis.com/org-bucket",
+                    filter="all",
+                    project_id=f"organizations/{org_id}",
+                    include_children=True,
+                ),
+            ]
+            logging_client.projects = {
+                GCP_PROJECT_ID: GCPProject(
+                    id=GCP_PROJECT_ID,
+                    number="123456789012",
+                    name="test",
+                    labels={},
+                    lifecycle_state="ACTIVE",
+                    organization=GCPOrganization(
+                        id=org_id, name=f"organizations/{org_id}"
+                    ),
+                )
+            }
+
+            check = logging_sink_created()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "PASS"
+            assert (
+                result[0].status_extended
+                == f"Sink project-sink is enabled exporting copies of all the log entries in project {GCP_PROJECT_ID}."
+            )
+            assert result[0].resource_id == "project-sink"
+            assert result[0].project_id == GCP_PROJECT_ID