fix(sdk): skip undescribed ECS task definitions (#12217)

Co-authored-by: Nguyễn Công Thuận Huy <nguyencongthuanhuy@gmail.com>
Co-authored-by: Hugo P.Brito <hugopbrit@gmail.com>

prowler/changelog.d/ecs-undescribed-task-definitions.fixed.md

@@ -0,0 +1 @@
+ECS task-definition checks no longer report PASS when `DescribeTaskDefinition` fails before container evidence is gathered

prowler/providers/aws/services/ecs/ecs_service.py

@@ -149,8 +149,8 @@ class ECS(AWSService):
                     "TAGS",
                 ],
             )
-            container_definitions = response["taskDefinition"]["containerDefinitions"]
-            for container in container_definitions:
+            container_definitions = []
+            for container in response["taskDefinition"]["containerDefinitions"]:
                 environment = []
                 if "environment" in container:
                     for env_var in container["environment"]:
@@ -159,7 +159,7 @@ class ECS(AWSService):
                                 name=env_var["name"], value=env_var["value"]
                             )
                         )
-                task_definition.container_definitions.append(
+                container_definitions.append(
                     ContainerDefinition(
                         name=container["name"],
                         privileged=container.get("privileged", False),
@@ -176,14 +176,16 @@ class ECS(AWSService):
                         .get("mode", ""),
                     )
                 )
-            task_definition.pid_mode = response["taskDefinition"].get("pidMode", "")
-            task_definition.registered_at = response["taskDefinition"].get(
-                "registeredAt"
-            )
-            task_definition.tags = response.get("tags")
-            task_definition.network_mode = response["taskDefinition"].get(
-                "networkMode", "bridge"
-            )
+            pid_mode = response["taskDefinition"].get("pidMode", "")
+            registered_at = response["taskDefinition"].get("registeredAt")
+            tags = response.get("tags")
+            network_mode = response["taskDefinition"].get("networkMode", "bridge")
+
+            task_definition.container_definitions = container_definitions
+            task_definition.pid_mode = pid_mode
+            task_definition.registered_at = registered_at
+            task_definition.tags = tags
+            task_definition.network_mode = network_mode
         except Exception as error:
             logger.error(
                 f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
@@ -302,7 +304,7 @@ class TaskDefinition(BaseModel):
     arn: str
     revision: str
     region: str
-    container_definitions: list[ContainerDefinition] = []
+    container_definitions: Optional[list[ContainerDefinition]] = None
     pid_mode: Optional[str]
     registered_at: Optional[datetime] = None
     tags: Optional[list] = []

prowler/providers/aws/services/ecs/ecs_task_definitions_containers_readonly_access/ecs_task_definitions_containers_readonly_access.py

@@ -6,6 +6,8 @@ class ecs_task_definitions_containers_readonly_access(Check):
     def execute(self):
         findings = []
         for task_definition in ecs_client.task_definitions.values():
+            if task_definition.container_definitions is None:
+                continue
             report = Check_Report_AWS(
                 metadata=self.metadata(), resource=task_definition
             )

prowler/providers/aws/services/ecs/ecs_task_definitions_host_namespace_not_shared/ecs_task_definitions_host_namespace_not_shared.py

@@ -6,6 +6,8 @@ class ecs_task_definitions_host_namespace_not_shared(Check):
     def execute(self):
         findings = []
         for task_definition in ecs_client.task_definitions.values():
+            if task_definition.container_definitions is None:
+                continue
             report = Check_Report_AWS(
                 metadata=self.metadata(), resource=task_definition
             )

prowler/providers/aws/services/ecs/ecs_task_definitions_host_networking_mode_users/ecs_task_definitions_host_networking_mode_users.py

@@ -6,6 +6,8 @@ class ecs_task_definitions_host_networking_mode_users(Check):
     def execute(self):
         findings = []
         for task_definition in ecs_client.task_definitions.values():
+            if task_definition.container_definitions is None:
+                continue
             report = Check_Report_AWS(
                 metadata=self.metadata(), resource=task_definition
             )

prowler/providers/aws/services/ecs/ecs_task_definitions_logging_block_mode/ecs_task_definitions_logging_block_mode.py

@@ -6,6 +6,8 @@ class ecs_task_definitions_logging_block_mode(Check):
     def execute(self):
         findings = []
         for task_definition in ecs_client.task_definitions.values():
+            if task_definition.container_definitions is None:
+                continue
             report = Check_Report_AWS(
                 metadata=self.metadata(), resource=task_definition
             )

prowler/providers/aws/services/ecs/ecs_task_definitions_logging_enabled/ecs_task_definitions_logging_enabled.py

@@ -6,6 +6,8 @@ class ecs_task_definitions_logging_enabled(Check):
     def execute(self):
         findings = []
         for task_definition in ecs_client.task_definitions.values():
+            if task_definition.container_definitions is None:
+                continue
             report = Check_Report_AWS(
                 metadata=self.metadata(), resource=task_definition
             )

prowler/providers/aws/services/ecs/ecs_task_definitions_no_environment_secrets/ecs_task_definitions_no_environment_secrets.py

@@ -16,7 +16,11 @@ class ecs_task_definitions_no_environment_secrets(Check):
             "secrets_ignore_patterns", []
         )
         validate = ecs_client.audit_config.get("secrets_validate", False)
-        task_definitions = list(ecs_client.task_definitions.values())
+        task_definitions = [
+            task_definition
+            for task_definition in ecs_client.task_definitions.values()
+            if task_definition.container_definitions is not None
+        ]
 
         # Scan every (task definition, container) environment in batched
         # Kingfisher invocations instead of one subprocess per container.

prowler/providers/aws/services/ecs/ecs_task_definitions_no_privileged_containers/ecs_task_definitions_no_privileged_containers.py

@@ -6,6 +6,8 @@ class ecs_task_definitions_no_privileged_containers(Check):
     def execute(self):
         findings = []
         for task_definition in ecs_client.task_definitions.values():
+            if task_definition.container_definitions is None:
+                continue
             report = Check_Report_AWS(
                 metadata=self.metadata(), resource=task_definition
             )

tests/providers/aws/services/ecs/ecs_task_definitions_undescribed_test.py

@@ -0,0 +1,164 @@
+from datetime import datetime, timezone
+from importlib import import_module
+from types import SimpleNamespace
+from unittest.mock import patch
+
+import botocore
+import pytest
+
+from prowler.providers.aws.services.ecs.ecs_service import ECS, TaskDefinition
+from tests.providers.aws.utils import (
+    AWS_ACCOUNT_NUMBER,
+    AWS_REGION_US_EAST_1,
+    set_mocked_aws_provider,
+)
+
+TASK_NAME = "test-task"
+TASK_REVISION = "1"
+TASK_ARN = (
+    f"arn:aws:ecs:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:"
+    f"task-definition/{TASK_NAME}:{TASK_REVISION}"
+)
+make_api_call = botocore.client.BaseClient._make_api_call
+
+
+def _mock_ecs_api(describe_result):
+    def mock_make_api_call(self, operation_name, kwargs):
+        if operation_name == "ListTaskDefinitions":
+            return {"taskDefinitionArns": [TASK_ARN]}
+        if operation_name == "DescribeTaskDefinition":
+            if isinstance(describe_result, Exception):
+                raise describe_result
+            return describe_result
+        if operation_name == "ListClusters":
+            return {"clusterArns": []}
+        return make_api_call(self, operation_name, kwargs)
+
+    return mock_make_api_call
+
+
+def _collect_task_definition(describe_result):
+    aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
+    with patch(
+        "botocore.client.BaseClient._make_api_call",
+        new=_mock_ecs_api(describe_result),
+    ):
+        return ECS(aws_provider).task_definitions[TASK_ARN]
+
+
+def _undescribed_ecs_client():
+    task_definition = TaskDefinition(
+        name=TASK_NAME,
+        arn=TASK_ARN,
+        revision=TASK_REVISION,
+        region=AWS_REGION_US_EAST_1,
+        environment_variables=[],
+    )
+    task_definition.container_definitions = None
+    return SimpleNamespace(
+        audit_config={},
+        task_definitions={TASK_ARN: task_definition},
+    )
+
+
+def test_failed_describe_leaves_task_definition_undescribed():
+    error = botocore.exceptions.ClientError(
+        {"Error": {"Code": "ThrottlingException", "Message": "rate exceeded"}},
+        "DescribeTaskDefinition",
+    )
+
+    task_definition = _collect_task_definition(error)
+
+    assert task_definition.container_definitions is None
+    assert task_definition.pid_mode is None
+    assert task_definition.network_mode is None
+
+
+def test_successful_describe_preserves_empty_container_definitions():
+    task_definition = _collect_task_definition(
+        {
+            "taskDefinition": {
+                "containerDefinitions": [],
+                "pidMode": "task",
+                "networkMode": "awsvpc",
+            },
+            "tags": [],
+        }
+    )
+
+    assert task_definition.container_definitions == []
+    assert task_definition.pid_mode == "task"
+    assert task_definition.network_mode == "awsvpc"
+
+
+def test_partial_parse_leaves_task_definition_undescribed():
+    task_definition = _collect_task_definition(
+        {
+            "taskDefinition": {
+                "containerDefinitions": [
+                    {"name": "valid-container"},
+                    {"privileged": False},
+                ],
+                "pidMode": "host",
+                "networkMode": "host",
+                "registeredAt": datetime(2026, 8, 13, tzinfo=timezone.utc),
+            },
+            "tags": [{"key": "Environment", "value": "production"}],
+        }
+    )
+
+    assert task_definition.container_definitions is None
+    assert task_definition.pid_mode is None
+    assert task_definition.network_mode is None
+    assert task_definition.registered_at is None
+    assert task_definition.tags == []
+
+
+@pytest.mark.parametrize(
+    ("check_package", "check_name"),
+    [
+        (
+            "ecs_task_definitions_containers_readonly_access",
+            "ecs_task_definitions_containers_readonly_access",
+        ),
+        (
+            "ecs_task_definitions_host_namespace_not_shared",
+            "ecs_task_definitions_host_namespace_not_shared",
+        ),
+        (
+            "ecs_task_definitions_host_networking_mode_users",
+            "ecs_task_definitions_host_networking_mode_users",
+        ),
+        (
+            "ecs_task_definitions_logging_block_mode",
+            "ecs_task_definitions_logging_block_mode",
+        ),
+        (
+            "ecs_task_definitions_logging_enabled",
+            "ecs_task_definitions_logging_enabled",
+        ),
+        (
+            "ecs_task_definitions_no_environment_secrets",
+            "ecs_task_definitions_no_environment_secrets",
+        ),
+        (
+            "ecs_task_definitions_no_privileged_containers",
+            "ecs_task_definitions_no_privileged_containers",
+        ),
+    ],
+)
+def test_undescribed_task_definitions_are_not_reported(
+    check_package, check_name, monkeypatch
+):
+    with patch(
+        "prowler.providers.common.provider.Provider.get_global_provider",
+        return_value=set_mocked_aws_provider([AWS_REGION_US_EAST_1]),
+    ):
+        module = import_module(
+            f"prowler.providers.aws.services.ecs.{check_package}.{check_name}"
+        )
+    monkeypatch.setattr(module, "ecs_client", _undescribed_ecs_client())
+
+    check = getattr(module, check_name)()
+
+    assert check.execute() == []