Back to Checks
gcp

Project has at least one logging sink exporting copies of all log entries

logging_sink_created
Service: loggingSeverity: medium
OfficialBuilt-inVerified
Categories:loggingforensics-ready
Resource Type:logging.googleapis.com/LogSink

Cloud Logging project contains at least one sink that exports a copy of all log entries to a destination for centralized retention or processing

Prowler logoProwlerprowler-provider-gcp

Risk

Without exporting all logs, audit evidence can expire or be altered in-project, reducing the availability and integrity of telemetry. This hinders threat detection and forensics, prevents cross-project correlation, and can let attacker actions evade scrutiny after log rotation or deletion.

Run this check with Prowler

Run in Prowler Cloud
Run
prowler gcp --checks logging_sink_created

Recommendation

Create a centralized export sink that routes all logs to a secured, durable, preferably immutable destination with extended retention. Apply least privilege to sink identities, separate duties by isolating destinations, use defense in depth (encryption, access controls), and monitor sink health for continuity.

Remediation

CLI

gcloud logging sinks create <example_resource_name> logging.googleapis.com/projects/<example_resource_id>

Terraform

# Create a project-level logging sink exporting all logs
resource "google_logging_project_sink" "sink" {
  name        = "<example_resource_name>" # critical: creates the sink required by the check
  destination = "logging.googleapis.com/projects/<example_resource_id>" # critical: required destination
  # No filter set -> exports all log entries (fixes the finding)
}

Other

1In Google Cloud Console, go to Logging > Log Router
2Click Create sink
3Set Sink name to <example_resource_name>
4For Sink destination, select Google Cloud project and choose <example_resource_id>
5Leave the inclusion filter empty (exports all logs)
6Click Create

Check Code

from prowler.lib.check.models import Check, Check_Report_GCP
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" 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:
            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(
                    metadata=self.metadata(),
                    resource=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} 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

References