feat(ecs): Add ecs_task_definitions_logging_block_mode check (#5526)

2026-07-24 13:01:56 +00:00 · 2024-10-30 13:02:36 -04:00
parent 056d482023
commit a2b6bdc461
6 changed files with 255 additions and 1 deletions
@@ -77,6 +77,9 @@ class ECS(AWSService):
                        log_driver=container.get("logConfiguration", {}).get(
                            "logDriver", ""
                        ),
+                        log_option=container.get("logConfiguration", {})
+                        .get("options", {})
+                        .get("mode", ""),
                    )
                )
            task_definition.pid_mode = response["taskDefinition"].get("pidMode", "")
@@ -180,6 +183,7 @@ class ContainerDefinition(BaseModel):
    user: str
    environment: list[ContainerEnvVariable]
    log_driver: Optional[str]
+    log_option: Optional[str]


 class TaskDefinition(BaseModel):
@@ -0,0 +1,34 @@
+{
+  "Provider": "aws",
+  "CheckID": "ecs_task_definitions_logging_block_mode",
+  "CheckTitle": "ECS task definitions containers should have a logging configured with non blocking mode",
+  "CheckType": [
+    "Resiliance"
+  ],
+  "ServiceName": "ecs",
+  "SubServiceName": "",
+  "ResourceIdTemplate": "arn:aws:ecs:{region}:{account-id}:task-definition/{task-definition-name}",
+  "Severity": "low",
+  "ResourceType": "AwsEcsTaskDefinition",
+  "Description": "This control checks if the latest active Amazon ECS task definition has a logging configuration set to non blocking mode.",
+  "Risk": "When logs cannot be immediately sent to Amazon CloudWatch, calls from container code to write to stdout or stderr will block and halt execution of the code. The logging thread in the application will block, which may prevent the application from functioning and lead to health check failures and task termination. Container startup fails if the required log group or log stream cannot be created.",
+  "RelatedUrl": "https://docs.aws.amazon.com/config/latest/developerguide/ecs-task-definition-log-configuration.html",
+  "Remediation": {
+    "Code": {
+      "CLI": "aws ecs register-task-definition --family <task-family> --container-definitions '[{\"name\":\"<container-name>\",\"image\":\"<image>\",\"logConfiguration\":{\"logDriver\":\"awslogs\",\"options\":{\"awslogs-group\":\"<log-group>\",\"awslogs-region\":\"<region>\",\"awslogs-stream-prefix\":\"ecs\",\"mode\":\"non-blocking\"}}}]'",
+      "NativeIaC": "",
+      "Other": "https://docs.aws.amazon.com/awssupport/latest/user/fault-tolerance-checks.html#amazon-ec2-awslogs-driver-blockingmode",
+      "Terraform": ""
+    },
+    "Recommendation": {
+      "Text": "Configure logging in ECS task definition to non blocking mode to ensure any issues writing logs will not block or halt the container execution.",
+      "Url": "https://docs.aws.amazon.com/AmazonECS/latest/developerguide/using_awslogs.html#specify-log-config"
+    }
+  },
+  "Categories": [
+    "logging"
+  ],
+  "DependsOn": [],
+  "RelatedTo": [],
+  "Notes": ""
+}
@@ -0,0 +1,30 @@
+from prowler.lib.check.models import Check, Check_Report_AWS
+from prowler.providers.aws.services.ecs.ecs_client import ecs_client
+
+
+class ecs_task_definitions_logging_block_mode(Check):
+    def execute(self):
+        findings = []
+        for task_definition in ecs_client.task_definitions.values():
+            report = Check_Report_AWS(self.metadata())
+            containers = 0
+            report.region = task_definition.region
+            report.resource_id = f"{task_definition.name}:{task_definition.revision}"
+            report.resource_arn = task_definition.arn
+            report.resource_tags = task_definition.tags
+            report.status = "PASS"
+            report.status_extended = f"ECS task definition {task_definition.name} with revision {task_definition.revision} containers has logging configured with non blocking mode."
+            failed_containers = []
+            for container in task_definition.container_definitions:
+                if container.log_driver:
+                    containers = containers + 1
+                    if container.log_option != "non-blocking":
+                        report.status = "FAIL"
+                        failed_containers.append(container.name)
+
+            if failed_containers:
+                report.status_extended = f"ECS task definition {task_definition.name} with revision {task_definition.revision} running with logging set to blocking mode on containers: {', '.join(failed_containers)}"
+
+            if containers > 0:
+                findings.append(report)
+        return findings
@@ -22,6 +22,13 @@ def mock_make_api_call(self, operation_name, kwarg):
                    {
                        "name": "test-container",
                        "image": "test-image",
+                        "logConfiguration": {
+                            "logDriver": "awslogs",
+                            "options": {
+                                "mode": "non-blocking",
+                                "max-buffer-size": "25m",
+                            },
+                        },
                        "environment": [
                            {"name": "DB_PASSWORD", "value": "pass-12343"},
                        ],
@@ -163,7 +170,14 @@ class Test_ECS_Service:
        assert ecs.task_definitions[task_arn].network_mode == "host"
        assert not ecs.task_definitions[task_arn].container_definitions[0].privileged
        assert ecs.task_definitions[task_arn].container_definitions[0].user == ""
-        assert ecs.task_definitions[task_arn].container_definitions[0].log_driver == ""
+        assert (
+            ecs.task_definitions[task_arn].container_definitions[0].log_driver
+            == "awslogs"
+        )
+        assert (
+            ecs.task_definitions[task_arn].container_definitions[0].log_option
+            == "non-blocking"
+        )
        assert ecs.task_definitions[task_arn].pid_mode == "host"
        assert (
            not ecs.task_definitions[task_arn]
@@ -0,0 +1,172 @@
+from unittest.mock import patch
+
+from boto3 import client
+from moto import mock_aws
+
+from tests.providers.aws.utils import AWS_REGION_US_EAST_1, set_mocked_aws_provider
+
+TASK_NAME = "test-task"
+TASK_REVISION = "1"
+CONTAINER_NAME = "test-container"
+
+
+class Test_ecs_task_definitions_logging_block_mode:
+    def test_no_task_definitions(self):
+        from prowler.providers.aws.services.ecs.ecs_service import ECS
+
+        mocked_aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
+
+        with patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=mocked_aws_provider,
+        ), patch(
+            "prowler.providers.aws.services.ecs.ecs_task_definitions_logging_block_mode.ecs_task_definitions_logging_block_mode.ecs_client",
+            new=ECS(mocked_aws_provider),
+        ):
+            from prowler.providers.aws.services.ecs.ecs_task_definitions_logging_block_mode.ecs_task_definitions_logging_block_mode import (
+                ecs_task_definitions_logging_block_mode,
+            )
+
+            check = ecs_task_definitions_logging_block_mode()
+            result = check.execute()
+            assert len(result) == 0
+
+    @mock_aws
+    def test_task_definition_no_logconfiguration(self):
+        ecs_client = client("ecs", region_name=AWS_REGION_US_EAST_1)
+
+        ecs_client.register_task_definition(
+            family=TASK_NAME,
+            containerDefinitions=[
+                {
+                    "name": CONTAINER_NAME,
+                    "image": "ubuntu",
+                    "memory": 128,
+                    "readonlyRootFilesystem": True,
+                    "privileged": False,
+                    "user": "appuser",
+                    "environment": [],
+                }
+            ],
+        )["taskDefinition"]["taskDefinitionArn"]
+
+        from prowler.providers.aws.services.ecs.ecs_service import ECS
+
+        mocked_aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
+
+        with patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=mocked_aws_provider,
+        ), patch(
+            "prowler.providers.aws.services.ecs.ecs_task_definitions_logging_block_mode.ecs_task_definitions_logging_block_mode.ecs_client",
+            new=ECS(mocked_aws_provider),
+        ):
+            from prowler.providers.aws.services.ecs.ecs_task_definitions_logging_block_mode.ecs_task_definitions_logging_block_mode import (
+                ecs_task_definitions_logging_block_mode,
+            )
+
+            check = ecs_task_definitions_logging_block_mode()
+            result = check.execute()
+            assert len(result) == 0
+
+    @mock_aws
+    def test_task_definition_log_configuration_no_mode(self):
+        ecs_client = client("ecs", region_name=AWS_REGION_US_EAST_1)
+
+        task_arn = ecs_client.register_task_definition(
+            family=TASK_NAME,
+            containerDefinitions=[
+                {
+                    "name": CONTAINER_NAME,
+                    "image": "ubuntu",
+                    "memory": 128,
+                    "readonlyRootFilesystem": True,
+                    "privileged": False,
+                    "user": "appuser",
+                    "environment": [],
+                    "logConfiguration": {"logDriver": "awslogs"},
+                }
+            ],
+        )["taskDefinition"]["taskDefinitionArn"]
+
+        from prowler.providers.aws.services.ecs.ecs_service import ECS
+
+        mocked_aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
+
+        with patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=mocked_aws_provider,
+        ), patch(
+            "prowler.providers.aws.services.ecs.ecs_task_definitions_logging_block_mode.ecs_task_definitions_logging_block_mode.ecs_client",
+            new=ECS(mocked_aws_provider),
+        ):
+            from prowler.providers.aws.services.ecs.ecs_task_definitions_logging_block_mode.ecs_task_definitions_logging_block_mode import (
+                ecs_task_definitions_logging_block_mode,
+            )
+
+            check = ecs_task_definitions_logging_block_mode()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert (
+                result[0].status_extended
+                == f"ECS task definition {TASK_NAME} with revision {TASK_REVISION} running with logging set to blocking mode on containers: {CONTAINER_NAME}"
+            )
+            assert result[0].resource_id == f"{TASK_NAME}:{TASK_REVISION}"
+            assert result[0].resource_arn == task_arn
+            assert result[0].region == AWS_REGION_US_EAST_1
+            assert result[0].resource_tags == []
+
+    @mock_aws
+    def test_task_definition_log_configuration_block_mode(self):
+        ecs_client = client("ecs", region_name=AWS_REGION_US_EAST_1)
+
+        task_arn = ecs_client.register_task_definition(
+            family=TASK_NAME,
+            containerDefinitions=[
+                {
+                    "name": CONTAINER_NAME,
+                    "image": "ubuntu",
+                    "memory": 128,
+                    "readonlyRootFilesystem": True,
+                    "privileged": True,
+                    "user": "root",
+                    "environment": [],
+                    "logConfiguration": {
+                        "logDriver": "awslogs",
+                        "options": {
+                            "mode": "non-blocking",
+                            "max-buffer-size": "25m",
+                        },
+                    },
+                }
+            ],
+        )["taskDefinition"]["taskDefinitionArn"]
+
+        from prowler.providers.aws.services.ecs.ecs_service import ECS
+
+        mocked_aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
+
+        with patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=mocked_aws_provider,
+        ), patch(
+            "prowler.providers.aws.services.ecs.ecs_task_definitions_logging_block_mode.ecs_task_definitions_logging_block_mode.ecs_client",
+            new=ECS(mocked_aws_provider),
+        ):
+            from prowler.providers.aws.services.ecs.ecs_task_definitions_logging_block_mode.ecs_task_definitions_logging_block_mode import (
+                ecs_task_definitions_logging_block_mode,
+            )
+
+            check = ecs_task_definitions_logging_block_mode()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "PASS"
+            assert (
+                result[0].status_extended
+                == f"ECS task definition {TASK_NAME} with revision {TASK_REVISION} containers has logging configured with non blocking mode."
+            )
+            assert result[0].resource_id == f"{TASK_NAME}:{TASK_REVISION}"
+            assert result[0].resource_arn == task_arn
+            assert result[0].region == AWS_REGION_US_EAST_1
+            assert result[0].resource_tags == []