feat(ecs): add new check ecs_task_definitions_host_networking_mode_users (#5088)

Co-authored-by: Sergio <sergio@prowler.com>
2026-07-24 21:11:53 +00:00 · 2024-09-19 00:28:31 +02:00
parent 3bea772c6b
commit 8a8c2b5097
8 changed files with 420 additions and 133 deletions
@@ -13,7 +13,7 @@ class ECS(AWSService):
    def __init__(self, provider):
        # Call AWSService's __init__
        super().__init__(__class__.__name__, provider)
-        self.task_definitions = []
+        self.task_definitions = {}
        self.__threading_call__(self._list_task_definitions)
        self._describe_task_definition()

@@ -26,15 +26,13 @@ class ECS(AWSService):
                    if not self.audit_resources or (
                        is_resource_filtered(task_definition, self.audit_resources)
                    ):
-                        self.task_definitions.append(
-                            TaskDefinition(
-                                # we want the family name without the revision
-                                name=sub(":.*", "", task_definition.split("/")[1]),
-                                arn=task_definition,
-                                revision=task_definition.split(":")[-1],
-                                region=regional_client.region,
-                                environment_variables=[],
-                            )
+                        self.task_definitions[task_definition] = TaskDefinition(
+                            # we want the family name without the revision
+                            name=sub(":.*", "", task_definition.split("/")[1]),
+                            arn=task_definition,
+                            revision=task_definition.split(":")[-1],
+                            region=regional_client.region,
+                            environment_variables=[],
                        )
        except Exception as error:
            logger.error(
@@ -44,7 +42,7 @@ class ECS(AWSService):
    def _describe_task_definition(self):
        logger.info("ECS - Describing Task Definitions...")
        try:
-            for task_definition in self.task_definitions:
+            for task_definition in self.task_definitions.values():
                client = self.regional_clients[task_definition.region]
                response = client.describe_task_definition(
                    taskDefinition=task_definition.arn,
@@ -56,13 +54,22 @@ class ECS(AWSService):
                    "containerDefinitions"
                ]
                for container in container_definitions:
+                    environment = []
                    if "environment" in container:
                        for env_var in container["environment"]:
-                            task_definition.environment_variables.append(
+                            environment.append(
                                ContainerEnvVariable(
                                    name=env_var["name"], value=env_var["value"]
                                )
                            )
+                    task_definition.container_definitions.append(
+                        ContainerDefinition(
+                            name=container["name"],
+                            privileged=container.get("privileged", False),
+                            user=container.get("user", ""),
+                            environment=environment,
+                        )
+                    )
                task_definition.tags = response.get("tags")
                task_definition.network_mode = response["taskDefinition"].get(
                    "networkMode"
@@ -78,11 +85,18 @@ class ContainerEnvVariable(BaseModel):
    value: str


+class ContainerDefinition(BaseModel):
+    name: str
+    privileged: bool
+    user: str
+    environment: list[ContainerEnvVariable]
+
+
 class TaskDefinition(BaseModel):
    name: str
    arn: str
    revision: str
    region: str
-    environment_variables: list[ContainerEnvVariable]
+    container_definitions: list[ContainerDefinition] = []
    tags: Optional[list] = []
    network_mode: Optional[str]
@@ -0,0 +1,34 @@
+{
+  "Provider": "aws",
+  "CheckID": "ecs_task_definitions_host_networking_mode_users",
+  "CheckTitle": "Amazon ECS task definitions should have secure networking modes and user definitions",
+  "CheckType": [
+    "Software and Configuration Checks/AWS Security Best Practices"
+  ],
+  "ServiceName": "ecs",
+  "SubServiceName": "",
+  "ResourceIdTemplate": "arn:aws:ecs:region:account-id:task-definition/resource-id",
+  "Severity": "high",
+  "ResourceType": "AwsEcsTaskDefinition",
+  "Description": "This control checks whether an active Amazon ECS task definition with host networking mode has privileged or user container definitions. The control fails for task definitions that have host network mode and container definitions of privileged=false or empty, and user=root or empty.",
+  "Risk": "If ECS tasks are configured with host networking and either lack a defined user or run with elevated privileges, this can lead to privilege escalation, unauthorized access to resources, and increased exposure to vulnerabilities.",
+  "RelatedUrl": "https://docs.aws.amazon.com/config/latest/developerguide/ecs-task-definition-user-for-host-mode-check.html",
+  "Remediation": {
+    "Code": {
+      "CLI": "aws ecs update-task-definition --task-definition <task-definition-name> --network-mode awsvpc --requires-compatibilities FARGATE --user <user-name>",
+      "NativeIaC": "",
+      "Other": "https://docs.aws.amazon.com/securityhub/latest/userguide/ecs-controls.html#ecs-6",
+      "Terraform": ""
+    },
+    "Recommendation": {
+      "Text": "Ensure your ECS task definitions use the awsvpc networking mode and have secure user configurations when using host networking mode.",
+      "Url": "https://docs.aws.amazon.com/AmazonECS/latest/developerguide/update-task-definition-console-v2.html"
+    }
+  },
+  "Categories": [
+    "trustboundaries"
+  ],
+  "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_host_networking_mode_users(Check):
+    def execute(self):
+        findings = []
+        for task_definition in ecs_client.task_definitions.values():
+            report = Check_Report_AWS(self.metadata())
+            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}' does not have host network mode."
+            failed_containers = []
+            if task_definition.network_mode == "host":
+                for container in task_definition.container_definitions:
+                    if not container.privileged and (
+                        container.user == "root" or container.user == ""
+                    ):
+                        report.status = "FAIL"
+                        failed_containers.append(container.name)
+
+                if failed_containers:
+                    report.status_extended = f"ECS task definition '{task_definition.name}' has containers with host network mode and non-privileged containers running as root or with no user specified: {', '.join(failed_containers)}"
+                else:
+                    report.status_extended = f"ECS task definition '{task_definition.name}' has host network mode but no containers running as root or with no user specified."
+            findings.append(report)
+        return findings
@@ -11,33 +11,48 @@ class ecs_task_definitions_no_environment_secrets(Check):
        secrets_ignore_patterns = ecs_client.audit_config.get(
            "secrets_ignore_patterns", []
        )
-        for task_definition in ecs_client.task_definitions:
+        for task_definition in ecs_client.task_definitions.values():
            report = Check_Report_AWS(self.metadata())
            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"No secrets found in variables of ECS task definition {task_definition.name} with revision {task_definition.revision}."
-            if task_definition.environment_variables:
-                dump_env_vars = {}
-                for env_var in task_definition.environment_variables:
-                    dump_env_vars.update({env_var.name: env_var.value})
+            extended_status_parts = []

-                env_data = dumps(dump_env_vars, indent=2)
-                detect_secrets_output = detect_secrets_scan(
-                    data=env_data, excluded_secrets=secrets_ignore_patterns
-                )
-                if detect_secrets_output:
-                    secrets_string = ", ".join(
-                        [
-                            f"{secret['type']} on line {secret['line_number']}"
-                            for secret in detect_secrets_output
-                        ]
+            for container in task_definition.container_definitions:
+                container_secrets_found = []
+
+                if container.environment:
+                    dump_env_vars = {}
+                    for env_var in container.environment:
+                        dump_env_vars.update({env_var.name: env_var.value})
+
+                    env_data = dumps(dump_env_vars, indent=2)
+                    detect_secrets_output = detect_secrets_scan(
+                        data=env_data, excluded_secrets=secrets_ignore_patterns
                    )
+                    if detect_secrets_output:
+                        secrets_string = ", ".join(
+                            [
+                                f"{secret['type']} on line {secret['line_number']}"
+                                for secret in detect_secrets_output
+                            ]
+                        )
+                        container_secrets_found.append(
+                            f"Secrets in container {container.name} -> {secrets_string}"
+                        )
+                if container_secrets_found:
                    report.status = "FAIL"
-                    report.status_extended = f"Potential secret found in variables of ECS task definition {task_definition.name} with revision {task_definition.revision} -> {secrets_string}."
-
+                    extended_status_parts.extend(container_secrets_found)
+            if report.status == "FAIL":
+                report.status_extended = (
+                    f"Potential secrets found in ECS task definition {task_definition.name} with revision {task_definition.revision}: "
+                    + "; ".join(extended_status_parts)
+                    + "."
+                )
+            else:
+                report.status_extended = f"No secrets found in variables of ECS task definition {task_definition.name} with revision {task_definition.revision}."
            findings.append(report)

        return findings
@@ -1,11 +1,38 @@
 from unittest.mock import patch

-from boto3 import client
-from moto import mock_aws
+import botocore

 from prowler.providers.aws.services.ecs.ecs_service import ECS
 from tests.providers.aws.utils import AWS_REGION_EU_WEST_1, set_mocked_aws_provider

+make_api_call = botocore.client.BaseClient._make_api_call
+
+
+def mock_make_api_call(self, operation_name, kwarg):
+    if operation_name == "ListTaskDefinitions":
+        return {
+            "taskDefinitionArns": [
+                "arn:aws:ecs:eu-west-1:123456789012:task-definition/test_ecs_task:1"
+            ]
+        }
+    if operation_name == "DescribeTaskDefinition":
+        return {
+            "taskDefinition": {
+                "containerDefinitions": [
+                    {
+                        "name": "test-container",
+                        "image": "test-image",
+                        "environment": [
+                            {"name": "DB_PASSWORD", "value": "pass-12343"},
+                        ],
+                    }
+                ],
+                "networkMode": "host",
+                "tags": [],
+            }
+        }
+    return make_api_call(self, operation_name, kwarg)
+

 def mock_generate_regional_clients(provider, service):
    regional_client = provider._session.current_session.client(
@@ -40,94 +67,49 @@ class Test_ECS_Service:
        assert ecs.session.__class__.__name__ == "Session"

    # Test list ECS task definitions
-    @mock_aws
+    @patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call)
    def test_list_task_definitions(self):
-        ecs_client = client("ecs", region_name=AWS_REGION_EU_WEST_1)
-
-        definition = dict(
-            family="test_ecs_task",
-            containerDefinitions=[
-                {
-                    "name": "hello_world",
-                    "image": "hello-world:latest",
-                    "memory": 400,
-                }
-            ],
-        )
-
-        task_definition = ecs_client.register_task_definition(**definition)
        aws_provider = set_mocked_aws_provider()
        ecs = ECS(aws_provider)

-        assert len(ecs.task_definitions) == 1
-        assert (
-            ecs.task_definitions[0].name == task_definition["taskDefinition"]["family"]
-        )
-        assert (
-            ecs.task_definitions[0].arn
-            == task_definition["taskDefinition"]["taskDefinitionArn"]
-        )
-        assert ecs.task_definitions[0].environment_variables == []
+        task_arn = "arn:aws:ecs:eu-west-1:123456789012:task-definition/test_ecs_task:1"

-    @mock_aws
+        assert len(ecs.task_definitions) == 1
+        assert ecs.task_definitions[task_arn].name == "test_ecs_task"
+        assert ecs.task_definitions[task_arn].arn == task_arn
+        assert ecs.task_definitions[task_arn].revision == "1"
+        assert ecs.task_definitions[task_arn].region == AWS_REGION_EU_WEST_1
+
+    @patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call)
    # Test describe ECS task definitions
    def test__describe_task_definitions__(self):
-        ecs_client = client("ecs", region_name=AWS_REGION_EU_WEST_1)
-
-        definition = dict(
-            family="test_ecs_task",
-            containerDefinitions=[
-                {
-                    "name": "hello_world",
-                    "image": "hello-world:latest",
-                    "memory": 400,
-                    "environment": [
-                        {"name": "test-env", "value": "test-env-value"},
-                        {"name": "test-env2", "value": "test-env-value2"},
-                    ],
-                }
-            ],
-            tags=[
-                {"key": "test", "value": "test"},
-            ],
-        )
-
-        task_definition = ecs_client.register_task_definition(**definition)
        aws_provider = set_mocked_aws_provider()
        ecs = ECS(aws_provider)

+        task_arn = "arn:aws:ecs:eu-west-1:123456789012:task-definition/test_ecs_task:1"
+
        assert len(ecs.task_definitions) == 1
+        assert ecs.task_definitions[task_arn].name == "test_ecs_task"
+        assert ecs.task_definitions[task_arn].arn == task_arn
+        assert ecs.task_definitions[task_arn].revision == "1"
+        assert ecs.task_definitions[task_arn].region == AWS_REGION_EU_WEST_1
+        assert len(ecs.task_definitions[task_arn].container_definitions) == 1
        assert (
-            ecs.task_definitions[0].name == task_definition["taskDefinition"]["family"]
-        )
-        assert ecs.task_definitions[0].tags == [
-            {"key": "test", "value": "test"},
-        ]
-        assert (
-            ecs.task_definitions[0].arn
-            == task_definition["taskDefinition"]["taskDefinitionArn"]
+            ecs.task_definitions[task_arn].container_definitions[0].name
+            == "test-container"
        )
        assert (
-            ecs.task_definitions[0].environment_variables[0].name
-            == task_definition["taskDefinition"]["containerDefinitions"][0][
-                "environment"
-            ][0]["name"]
+            len(ecs.task_definitions[task_arn].container_definitions[0].environment)
+            == 1
        )
        assert (
-            ecs.task_definitions[0].environment_variables[0].value
-            == task_definition["taskDefinition"]["containerDefinitions"][0][
-                "environment"
-            ][0]["value"]
+            ecs.task_definitions[task_arn].container_definitions[0].environment[0].name
+            == "DB_PASSWORD"
        )
        assert (
-            ecs.task_definitions[0].environment_variables[1].name
-            == task_definition["taskDefinition"]["containerDefinitions"][0][
-                "environment"
-            ][1]["name"]
-        )
-        assert (
-            ecs.task_definitions[0].environment_variables[1].value
-            == task_definition["taskDefinition"]["containerDefinitions"][0][
-                "environment"
-            ][1]["value"]
+            ecs.task_definitions[task_arn].container_definitions[0].environment[0].value
+            == "pass-12343"
        )
+        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 == ""
@@ -1,6 +1,7 @@
 from unittest import mock

 from prowler.providers.aws.services.ecs.ecs_service import (
+    ContainerDefinition,
    ContainerEnvVariable,
    TaskDefinition,
 )
@@ -9,6 +10,7 @@ AWS_REGION = "eu-west-1"
 AWS_ACCOUNT_NUMBER = "123456789012"
 task_name = "test-task"
 task_revision = "1"
+task_arn = f"arn:aws:ecs:{AWS_REGION}:{AWS_ACCOUNT_NUMBER}:task-definition/{task_name}:{task_revision}"
 env_var_name_no_secrets = "host"
 env_var_value_no_secrets = "localhost:1234"
 env_var_name_with_secrets = "DB_PASSWORD"
@@ -18,7 +20,7 @@ env_var_value_with_secrets = "pass-12343"
 class Test_ecs_task_definitions_no_environment_secrets:
    def test_no_task_definitions(self):
        ecs_client = mock.MagicMock
-        ecs_client.task_definitions = []
+        ecs_client.task_definitions = {}

        with mock.patch(
            "prowler.providers.aws.services.ecs.ecs_service.ECS",
@@ -34,19 +36,24 @@ class Test_ecs_task_definitions_no_environment_secrets:

    def test_container_env_var_no_secrets(self):
        ecs_client = mock.MagicMock
-        ecs_client.task_definitions = []
-        ecs_client.task_definitions.append(
-            TaskDefinition(
-                name=task_name,
-                arn=f"arn:aws:ecs:{AWS_REGION}:{AWS_ACCOUNT_NUMBER}:task-definition/{task_name}:{task_revision}",
-                revision="1",
-                region=AWS_REGION,
-                environment_variables=[
-                    ContainerEnvVariable(
-                        name=env_var_name_no_secrets, value=env_var_value_no_secrets
-                    )
-                ],
-            )
+        ecs_client.task_definitions = {}
+        ecs_client.task_definitions[task_arn] = TaskDefinition(
+            name=task_name,
+            arn=f"arn:aws:ecs:{AWS_REGION}:{AWS_ACCOUNT_NUMBER}:task-definition/{task_name}:{task_revision}",
+            revision="1",
+            region=AWS_REGION,
+            container_definitions=[
+                ContainerDefinition(
+                    name="container1",
+                    privileged=False,
+                    user="",
+                    environment=[
+                        ContainerEnvVariable(
+                            name=env_var_name_no_secrets, value=env_var_value_no_secrets
+                        )
+                    ],
+                )
+            ],
        )

        with mock.patch(
@@ -73,19 +80,25 @@ class Test_ecs_task_definitions_no_environment_secrets:

    def test_container_env_var_with_secrets(self):
        ecs_client = mock.MagicMock
-        ecs_client.task_definitions = []
-        ecs_client.task_definitions.append(
-            TaskDefinition(
-                name=task_name,
-                arn=f"arn:aws:ecs:{AWS_REGION}:{AWS_ACCOUNT_NUMBER}:task-definition/{task_name}:{task_revision}",
-                revision="1",
-                region=AWS_REGION,
-                environment_variables=[
-                    ContainerEnvVariable(
-                        name=env_var_name_with_secrets, value=env_var_value_with_secrets
-                    )
-                ],
-            )
+        ecs_client.task_definitions = {}
+        ecs_client.task_definitions[task_arn] = TaskDefinition(
+            name=task_name,
+            arn=f"arn:aws:ecs:{AWS_REGION}:{AWS_ACCOUNT_NUMBER}:task-definition/{task_name}:{task_revision}",
+            revision="1",
+            region=AWS_REGION,
+            container_definitions=[
+                ContainerDefinition(
+                    name="container1",
+                    privileged=False,
+                    user="",
+                    environment=[
+                        ContainerEnvVariable(
+                            name=env_var_name_with_secrets,
+                            value=env_var_value_with_secrets,
+                        )
+                    ],
+                )
+            ],
        )

        with mock.patch(
@@ -102,7 +115,7 @@ class Test_ecs_task_definitions_no_environment_secrets:
            assert result[0].status == "FAIL"
            assert (
                result[0].status_extended
-                == f"Potential secret found in variables of ECS task definition {task_name} with revision {task_revision} -> Secret Keyword on line 2."
+                == f"Potential secrets found in ECS task definition {task_name} with revision {task_revision}: Secrets in container container1 -> Secret Keyword on line 2."
            )
            assert result[0].resource_id == f"{task_name}:1"
            assert (
@@ -0,0 +1,199 @@
+from unittest import mock
+
+from prowler.providers.aws.services.ecs.ecs_service import (
+    ContainerDefinition,
+    ContainerEnvVariable,
+    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"
+container_name = "test-container"
+task_arn = f"arn:aws:ecs:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:task-definition/{task_name}:{task_revision}"
+
+
+class Test_ecs_task_definitions_host_networking_mode_users:
+    def test_no_task_definitions(self):
+        ecs_client = mock.MagicMock
+        ecs_client.task_definitions = {}
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=set_mocked_aws_provider([AWS_REGION_US_EAST_1]),
+        ), mock.patch(
+            "prowler.providers.aws.services.ecs.ecs_service.ECS",
+            ecs_client,
+        ):
+            from prowler.providers.aws.services.ecs.ecs_task_definitions_host_networking_mode_users.ecs_task_definitions_host_networking_mode_users import (
+                ecs_task_definitions_host_networking_mode_users,
+            )
+
+            check = ecs_task_definitions_host_networking_mode_users()
+            result = check.execute()
+            assert len(result) == 0
+
+    def test_task_definition_no_host_network_mode(self):
+        ecs_client = mock.MagicMock
+        ecs_client.task_definitions = {}
+        ecs_client.task_definitions[task_arn] = TaskDefinition(
+            name=task_name,
+            arn=task_arn,
+            revision=task_revision,
+            region=AWS_REGION_US_EAST_1,
+            network_mode="bridge",
+            container_definitions=[
+                ContainerDefinition(
+                    name=container_name,
+                    privileged=False,
+                    user="",
+                    environment=[
+                        ContainerEnvVariable(
+                            name="env_var_name_no_secrets",
+                            value="env_var_value_no_secrets",
+                        )
+                    ],
+                )
+            ],
+        )
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=set_mocked_aws_provider([AWS_REGION_US_EAST_1]),
+        ), mock.patch(
+            "prowler.providers.aws.services.ecs.ecs_service.ECS",
+            ecs_client,
+        ):
+            from prowler.providers.aws.services.ecs.ecs_task_definitions_host_networking_mode_users.ecs_task_definitions_host_networking_mode_users import (
+                ecs_task_definitions_host_networking_mode_users,
+            )
+
+            check = ecs_task_definitions_host_networking_mode_users()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "PASS"
+            assert (
+                result[0].status_extended
+                == f"ECS task definition '{task_name}' does not have host network mode."
+            )
+
+    def test_task_definition_host_mode_container_root_non_privileged(self):
+        ecs_client = mock.MagicMock
+        ecs_client.task_definitions = {}
+        ecs_client.task_definitions[task_arn] = TaskDefinition(
+            name=task_name,
+            arn=task_arn,
+            revision=task_revision,
+            region=AWS_REGION_US_EAST_1,
+            network_mode="host",
+            container_definitions=[
+                ContainerDefinition(
+                    name=container_name,
+                    privileged=False,
+                    user="root",
+                    environment=[],
+                )
+            ],
+        )
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=set_mocked_aws_provider([AWS_REGION_US_EAST_1]),
+        ), mock.patch(
+            "prowler.providers.aws.services.ecs.ecs_service.ECS",
+            ecs_client,
+        ):
+            from prowler.providers.aws.services.ecs.ecs_task_definitions_host_networking_mode_users.ecs_task_definitions_host_networking_mode_users import (
+                ecs_task_definitions_host_networking_mode_users,
+            )
+
+            check = ecs_task_definitions_host_networking_mode_users()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert (
+                result[0].status_extended
+                == f"ECS task definition '{task_name}' has containers with host network mode and non-privileged containers running as root or with no user specified: {container_name}"
+            )
+
+    def test_task_definition_host_mode_container_privileged(self):
+        ecs_client = mock.MagicMock
+        ecs_client.task_definitions = {}
+        ecs_client.task_definitions[task_arn] = TaskDefinition(
+            name=task_name,
+            arn=f"arn:aws:ecs:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:task-definition/{task_name}:{task_revision}",
+            revision=task_revision,
+            region=AWS_REGION_US_EAST_1,
+            network_mode="host",
+            container_definitions=[
+                ContainerDefinition(
+                    name=container_name,
+                    privileged=True,
+                    user="root",
+                    environment=[],
+                )
+            ],
+        )
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=set_mocked_aws_provider([AWS_REGION_US_EAST_1]),
+        ), mock.patch(
+            "prowler.providers.aws.services.ecs.ecs_service.ECS",
+            ecs_client,
+        ):
+            from prowler.providers.aws.services.ecs.ecs_task_definitions_host_networking_mode_users.ecs_task_definitions_host_networking_mode_users import (
+                ecs_task_definitions_host_networking_mode_users,
+            )
+
+            check = ecs_task_definitions_host_networking_mode_users()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "PASS"
+            assert (
+                result[0].status_extended
+                == f"ECS task definition '{task_name}' has host network mode but no containers running as root or with no user specified."
+            )
+
+    def test_task_definition_host_mode_container_not_root(self):
+        ecs_client = mock.MagicMock
+        ecs_client.task_definitions = {}
+        ecs_client.task_definitions[task_arn] = TaskDefinition(
+            name=task_name,
+            arn=task_arn,
+            revision=task_revision,
+            region=AWS_REGION_US_EAST_1,
+            network_mode="host",
+            container_definitions=[
+                ContainerDefinition(
+                    name=container_name,
+                    privileged=False,
+                    user="appuser",
+                    environment=[],
+                )
+            ],
+        )
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=set_mocked_aws_provider([AWS_REGION_US_EAST_1]),
+        ), mock.patch(
+            "prowler.providers.aws.services.ecs.ecs_service.ECS",
+            ecs_client,
+        ):
+            from prowler.providers.aws.services.ecs.ecs_task_definitions_host_networking_mode_users.ecs_task_definitions_host_networking_mode_users import (
+                ecs_task_definitions_host_networking_mode_users,
+            )
+
+            check = ecs_task_definitions_host_networking_mode_users()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "PASS"
+            assert (
+                result[0].status_extended
+                == f"ECS task definition '{task_name}' has host network mode but no containers running as root or with no user specified."
+            )