feat(aws): add new check cloudwatch_log_group_not_publicly_accessible (#5495)

Co-authored-by: Rubén De la Torre Vico <rubendltv22@gmail.com>
2026-07-24 21:11:53 +00:00 · 2024-10-30 08:50:17 -05:00
parent b87e6d20d7
commit 01716d9020
5 changed files with 299 additions and 0 deletions
@@ -0,0 +1,34 @@
+{
+  "Provider": "aws",
+  "CheckID": "cloudwatch_log_group_not_publicly_accessible",
+  "CheckTitle": "Ensure that CloudWatch Log Groups are not publicly accessible",
+  "CheckType": [
+    "Software and Configuration Checks/AWS Security Best Practices"
+  ],
+  "ServiceName": "cloudwatch",
+  "SubServiceName": "",
+  "ResourceIdTemplate": "arn:aws:ses:region:account-id:log-group/log_group_name",
+  "Severity": "high",
+  "ResourceType": "Other",
+  "Description": "This check ensures that no CloudWatch Log Groups are publicly accessible by checking for resource policies that allow access from any entity (Principal: '*'). Publicly exposed log groups pose a serious security risk as sensitive log data could be accessed by unauthorized parties.",
+  "Risk": "Publicly accessible CloudWatch Log Groups can expose sensitive information, leading to data breaches or unauthorized access. It is important to ensure that log groups are only accessible by trusted entities.",
+  "RelatedUrl": "https://docs.aws.amazon.com/AmazonCloudWatch/latest/logs/iam-access-control-overview-cwl.html",
+  "Remediation": {
+    "Code": {
+      "CLI": "aws logs delete-resource-policy --policy-name <policy-name>",
+      "NativeIaC": "",
+      "Other": "",
+      "Terraform": ""
+    },
+    "Recommendation": {
+      "Text": "Ensure that CloudWatch Log Groups are not publicly accessible. Review and remove any resource policies that allow public access (Principal: '*') to log groups.",
+      "Url": "https://docs.aws.amazon.com/AmazonCloudWatch/latest/logs/iam-access-control-overview-cwl.html"
+    }
+  },
+  "Categories": [
+    "internet-exposed"
+  ],
+  "DependsOn": [],
+  "RelatedTo": [],
+  "Notes": ""
+}
@@ -0,0 +1,45 @@
+from prowler.lib.check.models import Check, Check_Report_AWS
+from prowler.providers.aws.services.cloudwatch.logs_client import logs_client
+from prowler.providers.aws.services.iam.lib.policy import is_policy_public
+
+
+class cloudwatch_log_group_not_publicly_accessible(Check):
+    def execute(self):
+        findings = []
+        public_log_groups = []
+        if (
+            logs_client.resource_policies is not None
+            and logs_client.log_groups is not None
+        ):
+            for resource_policies in logs_client.resource_policies.values():
+                for resource_policy in resource_policies:
+                    if is_policy_public(
+                        resource_policy.policy, logs_client.audited_account
+                    ):
+                        for statement in resource_policy.policy.get("Statement", []):
+                            public_resources = statement.get("Resource", [])
+                            if isinstance(public_resources, str):
+                                public_resources = [public_resources]
+                            for resource in public_resources:
+                                for log_group in logs_client.log_groups.values():
+                                    if log_group.arn in resource or resource == "*":
+                                        public_log_groups.append(log_group.arn)
+            for log_group in logs_client.log_groups.values():
+                report = Check_Report_AWS(self.metadata())
+                report.region = log_group.region
+                report.resource_id = log_group.name
+                report.resource_arn = log_group.arn
+                report.resource_tags = log_group.tags
+                report.status = "PASS"
+                report.status_extended = (
+                    f"Log Group {log_group.name} is not publicly accessible."
+                )
+                if log_group.arn in public_log_groups:
+                    report.status = "FAIL"
+                    report.status_extended = (
+                        f"Log Group {log_group.name} is publicly accessible."
+                    )
+
+                findings.append(report)
+
+        return findings
@@ -1,3 +1,4 @@
+import json
 from datetime import datetime, timezone
 from typing import Optional

@@ -84,6 +85,8 @@ class Logs(AWSService):
        self.log_group_arn_template = f"arn:{self.audited_partition}:logs:{self.region}:{self.audited_account}:log-group"
        self.log_groups = {}
        self.__threading_call__(self._describe_log_groups)
+        self.resource_policies = {}
+        self.__threading_call__(self._describe_resource_policies)
        self.metric_filters = []
        self.__threading_call__(self._describe_metric_filters)
        if self.log_groups:
@@ -221,6 +224,38 @@ class Logs(AWSService):
            f"CloudWatch Logs - Finished retrieving log events in {regional_client.region}..."
        )

+    def _describe_resource_policies(self, regional_client):
+        logger.info("CloudWatch Logs - Describing resource policies...")
+        try:
+            describe_resource_policies_paginator = regional_client.get_paginator(
+                "describe_resource_policies"
+            )
+            if regional_client.region not in self.resource_policies:
+                self.resource_policies[regional_client.region] = []
+            for page in describe_resource_policies_paginator.paginate():
+                for policy in page["resourcePolicies"]:
+                    self.resource_policies[regional_client.region].append(
+                        ResourcePolicy(
+                            name=policy["policyName"],
+                            policy=json.loads(policy["policyDocument"]),
+                            region=regional_client.region,
+                        )
+                    )
+        except ClientError as error:
+            if error.response["Error"]["Code"] == "AccessDeniedException":
+                logger.error(
+                    f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+                )
+                self.resource_policies[regional_client.region] = None
+            else:
+                logger.error(
+                    f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+                )
+        except Exception as error:
+            logger.error(
+                f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+            )
+
    def _list_tags_for_resource(self, log_group):
        logger.info(f"CloudWatch Logs - List Tags for Log Group {log_group.name}...")
        try:
@@ -264,6 +299,12 @@ class LogGroup(BaseModel):
    tags: Optional[list] = []


+class ResourcePolicy(BaseModel):
+    name: str
+    policy: dict
+    region: str
+
+
 class MetricFilter(BaseModel):
    arn: str
    name: str
@@ -0,0 +1,179 @@
+import json
+from unittest import mock
+
+from boto3 import client
+from moto import mock_aws
+
+from tests.providers.aws.utils import AWS_REGION_US_EAST_1, set_mocked_aws_provider
+
+
+class Test_cloudwatch_log_group_not_publicly_accessible:
+    @mock_aws
+    def test_no_log_groups(self):
+        from prowler.providers.aws.services.cloudwatch.cloudwatch_service import Logs
+
+        aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=aws_provider,
+        ), mock.patch(
+            "prowler.providers.aws.services.cloudwatch.cloudwatch_log_group_not_publicly_accessible.cloudwatch_log_group_not_publicly_accessible.logs_client",
+            new=Logs(aws_provider),
+        ):
+            # Test Check
+            from prowler.providers.aws.services.cloudwatch.cloudwatch_log_group_not_publicly_accessible.cloudwatch_log_group_not_publicly_accessible import (
+                cloudwatch_log_group_not_publicly_accessible,
+            )
+
+            check = cloudwatch_log_group_not_publicly_accessible()
+            result = check.execute()
+
+            assert len(result) == 0
+
+    @mock_aws
+    def test_log_group_not_publicly_accessible(self):
+        # Generate Logs Client
+        logs_client = client("logs", region_name=AWS_REGION_US_EAST_1)
+        # Create Log Group without a public policy
+        logs_client.create_log_group(
+            logGroupName="test-log-group", tags={"test": "test"}
+        )
+
+        from prowler.providers.aws.services.cloudwatch.cloudwatch_service import Logs
+
+        aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=aws_provider,
+        ), mock.patch(
+            "prowler.providers.aws.services.cloudwatch.cloudwatch_log_group_not_publicly_accessible.cloudwatch_log_group_not_publicly_accessible.logs_client",
+            new=Logs(aws_provider),
+        ):
+            # Test Check
+            from prowler.providers.aws.services.cloudwatch.cloudwatch_log_group_not_publicly_accessible.cloudwatch_log_group_not_publicly_accessible import (
+                cloudwatch_log_group_not_publicly_accessible,
+            )
+
+            check = cloudwatch_log_group_not_publicly_accessible()
+            result = check.execute()
+
+            assert len(result) == 1
+            assert result[0].status == "PASS"
+            assert (
+                result[0].status_extended
+                == "Log Group test-log-group is not publicly accessible."
+            )
+            assert result[0].resource_id == "test-log-group"
+            assert (
+                result[0].resource_arn
+                == f"arn:aws:logs:{AWS_REGION_US_EAST_1}:123456789012:log-group:test-log-group"
+            )
+
+    @mock_aws
+    def test_log_group_publicly_accessible(self):
+        # Generate Logs Client
+        logs_client = client("logs", region_name=AWS_REGION_US_EAST_1)
+        # Create Log Group with a public policy
+        logs_client.create_log_group(
+            logGroupName="test-log-group", tags={"test": "test"}
+        )
+        logs_client.put_resource_policy(
+            policyName="PublicAccessPolicy",
+            policyDocument=json.dumps(
+                {
+                    "Version": "2012-10-17",
+                    "Statement": [
+                        {
+                            "Effect": "Allow",
+                            "Principal": "*",
+                            "Action": "logs:*",
+                            "Resource": "*",
+                        }
+                    ],
+                }
+            ),
+        )
+
+        from prowler.providers.aws.services.cloudwatch.cloudwatch_service import Logs
+
+        aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=aws_provider,
+        ), mock.patch(
+            "prowler.providers.aws.services.cloudwatch.cloudwatch_log_group_not_publicly_accessible.cloudwatch_log_group_not_publicly_accessible.logs_client",
+            new=Logs(aws_provider),
+        ):
+            # Test Check
+            from prowler.providers.aws.services.cloudwatch.cloudwatch_log_group_not_publicly_accessible.cloudwatch_log_group_not_publicly_accessible import (
+                cloudwatch_log_group_not_publicly_accessible,
+            )
+
+            check = cloudwatch_log_group_not_publicly_accessible()
+            result = check.execute()
+
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert (
+                result[0].status_extended
+                == "Log Group test-log-group is publicly accessible."
+            )
+            assert result[0].resource_id == "test-log-group"
+            assert (
+                result[0].resource_arn
+                == f"arn:aws:logs:{AWS_REGION_US_EAST_1}:123456789012:log-group:test-log-group"
+            )
+
+    @mock_aws
+    def test_log_group_empty_principal(self):
+        # Generate Logs Client
+        logs_client = client("logs", region_name=AWS_REGION_US_EAST_1)
+        # Create Log Group with a policy missing 'Principal'
+        logs_client.create_log_group(
+            logGroupName="test-log-group", tags={"test": "test"}
+        )
+        logs_client.put_resource_policy(
+            policyName="LimitedAccessPolicy",
+            policyDocument=json.dumps(
+                {
+                    "Version": "2012-10-17",
+                    "Statement": [
+                        {"Effect": "Allow", "Action": "logs:*", "Resource": "*"}
+                    ],
+                }
+            ),
+        )
+
+        from prowler.providers.aws.services.cloudwatch.cloudwatch_service import Logs
+
+        aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=aws_provider,
+        ), mock.patch(
+            "prowler.providers.aws.services.cloudwatch.cloudwatch_log_group_not_publicly_accessible.cloudwatch_log_group_not_publicly_accessible.logs_client",
+            new=Logs(aws_provider),
+        ):
+            # Test Check
+            from prowler.providers.aws.services.cloudwatch.cloudwatch_log_group_not_publicly_accessible.cloudwatch_log_group_not_publicly_accessible import (
+                cloudwatch_log_group_not_publicly_accessible,
+            )
+
+            check = cloudwatch_log_group_not_publicly_accessible()
+            result = check.execute()
+
+            assert len(result) == 1
+            assert result[0].status == "PASS"
+            assert (
+                result[0].status_extended
+                == "Log Group test-log-group is not publicly accessible."
+            )
+            assert result[0].resource_id == "test-log-group"
+            assert (
+                result[0].resource_arn
+                == f"arn:aws:logs:{AWS_REGION_US_EAST_1}:123456789012:log-group:test-log-group"
+            )