feat(cloudfront): detect Standard Logging v2 via CloudWatch Log Delivery (#10090)

Co-authored-by: Pepe Fagoaga <pepe@prowler.com>
2026-07-24 13:01:56 +00:00 · 2026-03-25 05:09:21 -05:00
parent c752811666
commit 833f3779ef
6 changed files with 271 additions and 10 deletions
@@ -8,6 +8,7 @@ All notable changes to the **Prowler SDK** are documented in this file.

 - `apikeys_api_restricted_with_gemini_api` check for GCP provider [(#10280)](https://github.com/prowler-cloud/prowler/pull/10280)
 - `gemini_api_disabled` check for GCP provider [(#10280)](https://github.com/prowler-cloud/prowler/pull/10280)
+- `cloudfront_distributions_logging_enabled` detects Standard Logging v2 via CloudWatch Log Delivery [(#10090)](https://github.com/prowler-cloud/prowler/pull/10090)

 ---

@@ -12,8 +12,8 @@
  "Severity": "medium",
  "ResourceType": "AwsCloudFrontDistribution",
  "ResourceGroup": "network",
-  "Description": "**CloudFront distributions** record viewer requests using either **standard access logs** or an attached **real-time log configuration**.\n\nThe finding evaluates whether logging is configured so request metadata is captured for each distribution.",
-  "Risk": "Missing **CloudFront logs** blinds monitoring of edge requests, impeding detection of bot abuse, credential stuffing, origin probing, and cache-bypass attempts.\n\nThis delays incident response and weakens evidence for forensics, impacting **confidentiality**, **integrity**, and **availability**.",
+  "Description": "**CloudFront distributions** record viewer requests using **standard access logs** (S3), **real-time log configurations**, or **Standard Logging v2** via CloudWatch Logs delivery sources.\n\nThe finding evaluates whether at least one logging mechanism is active so request metadata is captured for each distribution.",
+  "Risk": "Missing **CloudFront logs** blinds monitoring of edge requests, impeding detection of bot abuse, credential stuffing, and cache-bypass attempts.\n\nThis delays incident response and weakens forensic evidence. A delivery source without an active delivery does not count as enabled.",
  "RelatedUrl": "",
  "AdditionalURLs": [
    "https://docs.aws.amazon.com/AmazonCloudFront/latest/DeveloperGuide/AccessLogs.html",
@@ -9,14 +9,23 @@ class cloudfront_distributions_logging_enabled(Check):
        findings = []
        for distribution in cloudfront_client.distributions.values():
            report = Check_Report_AWS(metadata=self.metadata(), resource=distribution)
-            if distribution.logging_enabled or (
+            has_legacy_logging = distribution.logging_enabled
+            has_realtime_logging = (
                distribution.default_cache_config
                and distribution.default_cache_config.realtime_log_config_arn
-            ):
+            )
+            has_v2_logging = distribution.logging_v2_enabled
+
+            if has_legacy_logging or has_realtime_logging or has_v2_logging:
+                methods = []
+                if has_legacy_logging:
+                    methods.append("standard")
+                if has_realtime_logging:
+                    methods.append("real-time")
+                if has_v2_logging:
+                    methods.append("v2/CloudWatch")
                report.status = "PASS"
-                report.status_extended = (
-                    f"CloudFront Distribution {distribution.id} has logging enabled."
-                )
+                report.status_extended = f"CloudFront Distribution {distribution.id} has logging enabled via {', '.join(methods)}."
            else:
                report.status = "FAIL"
                report.status_extended = (
@@ -16,6 +16,7 @@ class CloudFront(AWSService):
        self._list_distributions(self.client, self.region)
        self._get_distribution_config(self.client, self.distributions, self.region)
        self._list_tags_for_resource(self.client, self.distributions, self.region)
+        self._get_log_delivery_sources(self.distributions, self.region)

    def _list_distributions(self, client, region) -> dict:
        logger.info("CloudFront - Listing Distributions...")
@@ -153,6 +154,54 @@ class CloudFront(AWSService):
                f"{region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
            )

+    def _get_log_delivery_sources(self, distributions, region):
+        """Check for Standard Logging v2 via CloudWatch Logs delivery sources.
+
+        A delivery source alone is not enough. We must also verify that an
+        active delivery exists for the source, otherwise logs are not flowing.
+        """
+        logger.info("CloudFront - Checking CloudWatch Logs delivery sources...")
+        try:
+            distribution_arns = {
+                dist.arn: dist_id for dist_id, dist in distributions.items()
+            }
+            if not distribution_arns:
+                return
+
+            # CloudFront delivery sources live in the global region (us-east-1),
+            # not the profile's default region.
+            global_region = self.provider.get_global_region()
+            logs_client = self.session.client("logs", region_name=global_region)
+
+            # Find delivery sources whose resourceArns match a distribution
+            matching_sources = {}
+            paginator = logs_client.get_paginator("describe_delivery_sources")
+            for page in paginator.paginate():
+                for source in page.get("deliverySources", []):
+                    if source.get("service") != self.service:
+                        continue
+                    for resource_arn in source.get("resourceArns", []):
+                        if resource_arn in distribution_arns:
+                            source_name = source.get("name", "")
+                            matching_sources[source_name] = resource_arn
+
+            if not matching_sources:
+                return
+
+            # Verify at least one active delivery exists per source
+            paginator = logs_client.get_paginator("describe_deliveries")
+            for page in paginator.paginate():
+                for delivery in page.get("deliveries", []):
+                    source_name = delivery.get("deliverySourceName", "")
+                    if source_name in matching_sources:
+                        resource_arn = matching_sources[source_name]
+                        dist_id = distribution_arns[resource_arn]
+                        distributions[dist_id].logging_v2_enabled = True
+        except Exception as error:
+            logger.error(
+                f"{region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+            )
+

 class OriginsSSLProtocols(Enum):
    SSLv3 = "SSLv3"
@@ -207,6 +256,7 @@ class Distribution(BaseModel):
    id: str
    region: str
    logging_enabled: bool = False
+    logging_v2_enabled: bool = False
    default_cache_config: Optional[DefaultCacheConfigBehaviour] = None
    geo_restriction_type: Optional[GeoRestrictionType] = None
    origins: list[Origin]
@@ -64,7 +64,7 @@ class Test_cloudfront_distributions_logging_enabled:
            assert result[0].status == "PASS"
            assert (
                result[0].status_extended
-                == f"CloudFront Distribution {DISTRIBUTION_ID} has logging enabled."
+                == f"CloudFront Distribution {DISTRIBUTION_ID} has logging enabled via standard."
            )

    def test_one_distribution_logging_disabled_realtime_disabled(self):
@@ -145,7 +145,7 @@ class Test_cloudfront_distributions_logging_enabled:
            assert result[0].status == "PASS"
            assert (
                result[0].status_extended
-                == f"CloudFront Distribution {DISTRIBUTION_ID} has logging enabled."
+                == f"CloudFront Distribution {DISTRIBUTION_ID} has logging enabled via real-time."
            )
            assert result[0].resource_tags == []

@@ -186,6 +186,48 @@ class Test_cloudfront_distributions_logging_enabled:
            assert result[0].status == "PASS"
            assert (
                result[0].status_extended
-                == f"CloudFront Distribution {DISTRIBUTION_ID} has logging enabled."
+                == f"CloudFront Distribution {DISTRIBUTION_ID} has logging enabled via standard, real-time."
+            )
+            assert result[0].resource_tags == []
+
+    def test_one_distribution_logging_v2_enabled(self):
+        cloudfront_client = mock.MagicMock
+        cloudfront_client.distributions = {
+            DISTRIBUTION_ID: Distribution(
+                arn=DISTRIBUTION_ARN,
+                id=DISTRIBUTION_ID,
+                region=REGION,
+                logging_enabled=False,
+                logging_v2_enabled=True,
+                default_cache_config=DefaultCacheConfigBehaviour(
+                    realtime_log_config_arn="",
+                    viewer_protocol_policy=ViewerProtocolPolicy.https_only,
+                    field_level_encryption_id="",
+                ),
+                origins=[],
+                origin_failover=False,
+            )
+        }
+
+        with mock.patch(
+            "prowler.providers.aws.services.cloudfront.cloudfront_service.CloudFront",
+            new=cloudfront_client,
+        ):
+            # Test Check
+            from prowler.providers.aws.services.cloudfront.cloudfront_distributions_logging_enabled.cloudfront_distributions_logging_enabled import (
+                cloudfront_distributions_logging_enabled,
+            )
+
+            check = cloudfront_distributions_logging_enabled()
+            result = check.execute()
+
+            assert len(result) == 1
+            assert result[0].region == REGION
+            assert result[0].resource_arn == DISTRIBUTION_ARN
+            assert result[0].resource_id == DISTRIBUTION_ID
+            assert result[0].status == "PASS"
+            assert (
+                result[0].status_extended
+                == f"CloudFront Distribution {DISTRIBUTION_ID} has logging enabled via v2/CloudWatch."
            )
            assert result[0].resource_tags == []
@@ -171,6 +171,10 @@ def mock_make_api_call(self, operation_name, kwarg):
                ]
            }
        }
+    if operation_name == "DescribeDeliverySources":
+        return {"deliverySources": []}
+    if operation_name == "DescribeDeliveries":
+        return {"deliveries": []}
    return make_api_call(self, operation_name, kwarg)


@@ -284,3 +288,158 @@ class Test_CloudFront_Service:
            == DEFAULT_CACHE_CONFIG.field_level_encryption_id
        )
        assert cloudfront.distributions[DISTRIBUTION_ID].tags == TAGS
+
+    def test_get_log_delivery_sources_with_active_delivery(self):
+        from tests.providers.aws.utils import AWS_ACCOUNT_NUMBER
+
+        DISTRIBUTION_ID = "E27LVI50CSW06W"
+        DISTRIBUTION_ARN = (
+            f"arn:aws:cloudfront::{AWS_ACCOUNT_NUMBER}:distribution/{DISTRIBUTION_ID}"
+        )
+        REGION = "us-east-1"
+
+        distributions = {
+            DISTRIBUTION_ID: Distribution(
+                arn=DISTRIBUTION_ARN,
+                id=DISTRIBUTION_ID,
+                region=REGION,
+                origins=[],
+                origin_failover=False,
+            )
+        }
+
+        mock_logs_client = mock.MagicMock()
+        mock_paginator_sources = mock.MagicMock()
+        mock_paginator_sources.paginate.return_value = [
+            {
+                "deliverySources": [
+                    {
+                        "name": "cf-source-1",
+                        "service": "cloudfront",
+                        "resourceArns": [DISTRIBUTION_ARN],
+                    },
+                ]
+            }
+        ]
+        mock_paginator_deliveries = mock.MagicMock()
+        mock_paginator_deliveries.paginate.return_value = [
+            {
+                "deliveries": [
+                    {
+                        "id": "delivery-1",
+                        "deliverySourceName": "cf-source-1",
+                        "deliveryDestinationArn": "arn:aws:logs:us-east-1:123456789012:log-group:cf-logs",
+                    },
+                ]
+            }
+        ]
+
+        def paginator_side_effect(operation):
+            if operation == "describe_delivery_sources":
+                return mock_paginator_sources
+            if operation == "describe_deliveries":
+                return mock_paginator_deliveries
+
+        mock_logs_client.get_paginator.side_effect = paginator_side_effect
+
+        service = mock.MagicMock()
+        service.service = "cloudfront"
+        service.session.client.return_value = mock_logs_client
+
+        CloudFront._get_log_delivery_sources(service, distributions, REGION)
+
+        assert distributions[DISTRIBUTION_ID].logging_v2_enabled is True
+
+    def test_get_log_delivery_sources_source_without_delivery(self):
+        from tests.providers.aws.utils import AWS_ACCOUNT_NUMBER
+
+        DISTRIBUTION_ID = "E27LVI50CSW06W"
+        DISTRIBUTION_ARN = (
+            f"arn:aws:cloudfront::{AWS_ACCOUNT_NUMBER}:distribution/{DISTRIBUTION_ID}"
+        )
+        REGION = "us-east-1"
+
+        distributions = {
+            DISTRIBUTION_ID: Distribution(
+                arn=DISTRIBUTION_ARN,
+                id=DISTRIBUTION_ID,
+                region=REGION,
+                origins=[],
+                origin_failover=False,
+            )
+        }
+
+        mock_logs_client = mock.MagicMock()
+        mock_paginator_sources = mock.MagicMock()
+        mock_paginator_sources.paginate.return_value = [
+            {
+                "deliverySources": [
+                    {
+                        "name": "cf-source-1",
+                        "service": "cloudfront",
+                        "resourceArns": [DISTRIBUTION_ARN],
+                    },
+                ]
+            }
+        ]
+        mock_paginator_deliveries = mock.MagicMock()
+        mock_paginator_deliveries.paginate.return_value = [{"deliveries": []}]
+
+        def paginator_side_effect(operation):
+            if operation == "describe_delivery_sources":
+                return mock_paginator_sources
+            if operation == "describe_deliveries":
+                return mock_paginator_deliveries
+
+        mock_logs_client.get_paginator.side_effect = paginator_side_effect
+
+        service = mock.MagicMock()
+        service.service = "cloudfront"
+        service.session.client.return_value = mock_logs_client
+
+        CloudFront._get_log_delivery_sources(service, distributions, REGION)
+
+        assert distributions[DISTRIBUTION_ID].logging_v2_enabled is False
+
+    def test_get_log_delivery_sources_no_matching_sources(self):
+        from tests.providers.aws.utils import AWS_ACCOUNT_NUMBER
+
+        DISTRIBUTION_ID = "E27LVI50CSW06W"
+        DISTRIBUTION_ARN = (
+            f"arn:aws:cloudfront::{AWS_ACCOUNT_NUMBER}:distribution/{DISTRIBUTION_ID}"
+        )
+        REGION = "us-east-1"
+
+        distributions = {
+            DISTRIBUTION_ID: Distribution(
+                arn=DISTRIBUTION_ARN,
+                id=DISTRIBUTION_ID,
+                region=REGION,
+                origins=[],
+                origin_failover=False,
+            )
+        }
+
+        mock_logs_client = mock.MagicMock()
+        mock_paginator_sources = mock.MagicMock()
+        mock_paginator_sources.paginate.return_value = [
+            {
+                "deliverySources": [
+                    {
+                        "name": "other-source",
+                        "service": "VPC",
+                        "resourceArns": ["arn:aws:some:other:resource"],
+                    },
+                ]
+            }
+        ]
+
+        mock_logs_client.get_paginator.return_value = mock_paginator_sources
+
+        service = mock.MagicMock()
+        service.service = "cloudfront"
+        service.session.client.return_value = mock_logs_client
+
+        CloudFront._get_log_delivery_sources(service, distributions, REGION)
+
+        assert distributions[DISTRIBUTION_ID].logging_v2_enabled is False