From 807f2f9f3021f3c808afee3cf32b0690e62c9645 Mon Sep 17 00:00:00 2001 From: Prowler Bot Date: Tue, 8 Oct 2024 22:34:53 +0200 Subject: [PATCH] chore(ecs): mock all tests using moto (#5333) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-authored-by: Rubén De la Torre Vico Co-authored-by: Sergio --- .../providers/aws/services/ecs/ecs_service.py | 1 - .../ecs_service_no_assign_public_ip_test.py | 185 ++++++++++--- ...nitions_containers_readonly_access_test.py | 198 ++++++++------ ...initions_host_namespace_not_shared_test.py | 211 ++++++++------- ...nitions_host_networking_mode_users_test.py | 242 +++++++++++------- ...s_task_definitions_logging_enabled_test.py | 151 +++++------ ...definitions_no_environment_secrets_test.py | 174 +++++++------ ...finitions_no_privileged_containers_test.py | 134 +++++----- 8 files changed, 759 insertions(+), 537 deletions(-) diff --git a/prowler/providers/aws/services/ecs/ecs_service.py b/prowler/providers/aws/services/ecs/ecs_service.py index 21d09e91ab..9a81df6377 100644 --- a/prowler/providers/aws/services/ecs/ecs_service.py +++ b/prowler/providers/aws/services/ecs/ecs_service.py @@ -8,7 +8,6 @@ from prowler.lib.scan_filters.scan_filters import is_resource_filtered from prowler.providers.aws.lib.service.service import AWSService -################################ ECS class ECS(AWSService): def __init__(self, provider): # Call AWSService's __init__ diff --git a/tests/providers/aws/services/ecs/ecs_service_no_assign_public_ip/ecs_service_no_assign_public_ip_test.py b/tests/providers/aws/services/ecs/ecs_service_no_assign_public_ip/ecs_service_no_assign_public_ip_test.py index f2fc01aa4a..38753ffeae 100644 --- a/tests/providers/aws/services/ecs/ecs_service_no_assign_public_ip/ecs_service_no_assign_public_ip_test.py +++ b/tests/providers/aws/services/ecs/ecs_service_no_assign_public_ip/ecs_service_no_assign_public_ip_test.py @@ -1,22 +1,87 @@ -from unittest import mock +from unittest.mock import patch -from prowler.providers.aws.services.ecs.ecs_service import Service -from tests.providers.aws.utils import AWS_ACCOUNT_NUMBER, AWS_REGION_US_EAST_1 +import botocore +from boto3 import client +from moto import mock_aws -SERVICE_ARN = ( - f"arn:aws:ecs:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:service/sample-service" +from tests.providers.aws.utils import ( + AWS_ACCOUNT_NUMBER, + AWS_REGION_US_EAST_1, + set_mocked_aws_provider, ) -SERVICE_NAME = "sample-service" + +orig = botocore.client.BaseClient._make_api_call + + +def mock_make_api_call(self, operation_name, kwarg): + if operation_name == "DescribeServices": + if kwarg["services"] == [ + f"arn:aws:ecs:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:service/sample-cluster/service-with-no-public-ip" + ]: + return { + "services": [ + { + "serviceName": "test-latest-linux-service", + "clusterArn": f"arn:aws:ecs:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:cluster/sample-cluster", + "taskDefinition": "test-task", + "loadBalancers": [], + "serviceArn": f"arn:aws:ecs:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:service/sample-cluster/service-with-no-public-ip", + "desiredCount": 1, + "launchType": "FARGATE", + "platformVersion": "1.4.0", + "platformFamily": "Linux", + "networkConfiguration": { + "awsvpcConfiguration": { + "subnets": ["subnet-12345678"], + "securityGroups": ["sg-12345678"], + "assignPublicIp": "DISABLED", + }, + }, + "tags": [], + }, + ], + } + elif kwarg["services"] == [ + f"arn:aws:ecs:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:service/sample-cluster/service-with-public-ip" + ]: + return { + "services": [ + { + "serviceName": "test-latest-linux-service", + "clusterArn": f"arn:aws:ecs:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:cluster/sample-cluster", + "taskDefinition": "test-task", + "loadBalancers": [], + "serviceArn": f"arn:aws:ecs:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:service/sample-cluster/service-with-public-ip", + "desiredCount": 1, + "launchType": "FARGATE", + "platformVersion": "1.4.0", + "platformFamily": "Linux", + "networkConfiguration": { + "awsvpcConfiguration": { + "subnets": ["subnet-12345678"], + "securityGroups": ["sg-12345678"], + "assignPublicIp": "ENABLED", + }, + }, + "tags": [], + }, + ], + } + return orig(self, operation_name, kwarg) class Test_ecs_service_no_assign_public_ip: def test_no_services(self): - ecs_client = mock.MagicMock - ecs_client.services = {} + from prowler.providers.aws.services.ecs.ecs_service import ECS - with mock.patch( - "prowler.providers.aws.services.ecs.ecs_service.ECS", - ecs_client, + 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_service_no_assign_public_ip.ecs_service_no_assign_public_ip.ecs_client", + new=ECS(mocked_aws_provider), ): from prowler.providers.aws.services.ecs.ecs_service_no_assign_public_ip.ecs_service_no_assign_public_ip import ( ecs_service_no_assign_public_ip, @@ -26,20 +91,37 @@ class Test_ecs_service_no_assign_public_ip: result = check.execute() assert len(result) == 0 + @patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call) + @mock_aws def test_service_with_no_public_ip(self): - ecs_client = mock.MagicMock - ecs_client.services = {} - ecs_client.services[SERVICE_ARN] = Service( - name=SERVICE_NAME, - arn=SERVICE_ARN, - region=AWS_REGION_US_EAST_1, - assign_public_ip=False, - tags=[], - ) + ecs_client = client("ecs", region_name=AWS_REGION_US_EAST_1) - with mock.patch( - "prowler.providers.aws.services.ecs.ecs_service.ECS", - ecs_client, + ecs_client.create_cluster(clusterName="sample-cluster") + + service_arn = ecs_client.create_service( + cluster="sample-cluster", + serviceName="service-with-no-public-ip", + desiredCount=1, + launchType="FARGATE", + networkConfiguration={ + "awsvpcConfiguration": { + "subnets": ["subnet-123456"], + "securityGroups": ["sg-123456"], + "assignPublicIp": "DISABLED", + } + }, + )["service"]["serviceArn"] + + 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_service_no_assign_public_ip.ecs_service_no_assign_public_ip.ecs_client", + new=ECS(mocked_aws_provider), ): from prowler.providers.aws.services.ecs.ecs_service_no_assign_public_ip.ecs_service_no_assign_public_ip import ( ecs_service_no_assign_public_ip, @@ -51,25 +133,44 @@ class Test_ecs_service_no_assign_public_ip: assert result[0].status == "PASS" assert ( result[0].status_extended - == f"ECS Service {SERVICE_NAME} does not have automatic public IP assignment." + == "ECS Service service-with-no-public-ip does not have automatic public IP assignment." ) - assert result[0].resource_id == SERVICE_NAME - assert result[0].resource_arn == SERVICE_ARN + assert result[0].resource_id == "service-with-no-public-ip" + assert result[0].resource_arn == service_arn + assert result[0].resource_tags == [] + assert result[0].region == AWS_REGION_US_EAST_1 + @patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call) + @mock_aws def test_task_definition_no_host_network_mode(self): - ecs_client = mock.MagicMock - ecs_client.services = {} - ecs_client.services[SERVICE_ARN] = Service( - name=SERVICE_NAME, - arn=SERVICE_ARN, - region=AWS_REGION_US_EAST_1, - assign_public_ip=True, - tags=[], - ) + ecs_client = client("ecs", region_name=AWS_REGION_US_EAST_1) - with mock.patch( - "prowler.providers.aws.services.ecs.ecs_service.ECS", - ecs_client, + ecs_client.create_cluster(clusterName="sample-cluster") + + service_arn = ecs_client.create_service( + cluster="sample-cluster", + serviceName="service-with-public-ip", + desiredCount=1, + launchType="FARGATE", + networkConfiguration={ + "awsvpcConfiguration": { + "subnets": ["subnet-123456"], + "securityGroups": ["sg-123456"], + "assignPublicIp": "ENABLED", + } + }, + )["service"]["serviceArn"] + + 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_service_no_assign_public_ip.ecs_service_no_assign_public_ip.ecs_client", + new=ECS(mocked_aws_provider), ): from prowler.providers.aws.services.ecs.ecs_service_no_assign_public_ip.ecs_service_no_assign_public_ip import ( ecs_service_no_assign_public_ip, @@ -81,7 +182,9 @@ class Test_ecs_service_no_assign_public_ip: assert result[0].status == "FAIL" assert ( result[0].status_extended - == f"ECS Service {SERVICE_NAME} has automatic public IP assignment." + == "ECS Service service-with-public-ip has automatic public IP assignment." ) - assert result[0].resource_id == SERVICE_NAME - assert result[0].resource_arn == SERVICE_ARN + assert result[0].resource_id == "service-with-public-ip" + assert result[0].resource_arn == service_arn + assert result[0].resource_tags == [] + assert result[0].region == AWS_REGION_US_EAST_1 diff --git a/tests/providers/aws/services/ecs/ecs_task_definitions_containers_readonly_access/ecs_task_definitions_containers_readonly_access_test.py b/tests/providers/aws/services/ecs/ecs_task_definitions_containers_readonly_access/ecs_task_definitions_containers_readonly_access_test.py index 13417518dd..a162ef87d8 100644 --- a/tests/providers/aws/services/ecs/ecs_task_definitions_containers_readonly_access/ecs_task_definitions_containers_readonly_access_test.py +++ b/tests/providers/aws/services/ecs/ecs_task_definitions_containers_readonly_access/ecs_task_definitions_containers_readonly_access_test.py @@ -1,10 +1,13 @@ -from unittest import mock +from unittest.mock import patch -from prowler.providers.aws.services.ecs.ecs_service import ( - ContainerDefinition, - TaskDefinition, +from boto3 import client +from moto import mock_aws + +from tests.providers.aws.utils import ( + AWS_ACCOUNT_NUMBER, + AWS_REGION_US_EAST_1, + set_mocked_aws_provider, ) -from tests.providers.aws.utils import AWS_ACCOUNT_NUMBER, AWS_REGION_US_EAST_1 TASK_NAME = "test-task-readonly" TASK_REVISION = "1" @@ -14,12 +17,16 @@ TASK_ARN = f"arn:aws:ecs:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:task-defini class Test_ecs_task_definitions_containers_readonly_access: def test_no_task_definitions(self): - ecs_client = mock.MagicMock - ecs_client.task_definitions = {} + from prowler.providers.aws.services.ecs.ecs_service import ECS - with mock.patch( - "prowler.providers.aws.services.ecs.ecs_service.ECS", - ecs_client, + 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_containers_readonly_access.ecs_task_definitions_containers_readonly_access.ecs_client", + new=ECS(mocked_aws_provider), ): from prowler.providers.aws.services.ecs.ecs_task_definitions_containers_readonly_access.ecs_task_definitions_containers_readonly_access import ( ecs_task_definitions_containers_readonly_access, @@ -29,29 +36,35 @@ class Test_ecs_task_definitions_containers_readonly_access: result = check.execute() assert len(result) == 0 + @mock_aws def test_task_definition_all_containers_readonly(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, - readonly_rootfilesystem=True, - privileged=False, - user="appuser", - environment=[], - ) - ], - ) + ecs_client = client("ecs", region_name=AWS_REGION_US_EAST_1) - with mock.patch( - "prowler.providers.aws.services.ecs.ecs_service.ECS", - ecs_client, + task_definition_arn = 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_containers_readonly_access.ecs_task_definitions_containers_readonly_access.ecs_client", + new=ECS(mocked_aws_provider), ): from prowler.providers.aws.services.ecs.ecs_task_definitions_containers_readonly_access.ecs_task_definitions_containers_readonly_access import ( ecs_task_definitions_containers_readonly_access, @@ -65,30 +78,40 @@ class Test_ecs_task_definitions_containers_readonly_access: result[0].status_extended == f"ECS task definition {TASK_NAME} with revision {TASK_REVISION} does not have containers with write access to the root filesystems." ) + assert result[0].resource_id == f"{TASK_NAME}:{TASK_REVISION}" + assert result[0].resource_arn == task_definition_arn + assert result[0].region == AWS_REGION_US_EAST_1 + assert result[0].resource_tags == [] + @mock_aws def test_task_definition_some_containers_not_readonly(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, - readonly_rootfilesystem=False, - privileged=False, - user="appuser", - environment=[], - ) + ecs_client = client("ecs", region_name=AWS_REGION_US_EAST_1) + + task_definition_arn = ecs_client.register_task_definition( + family=TASK_NAME, + containerDefinitions=[ + { + "name": CONTAINER_NAME, + "image": "ubuntu", + "memory": 128, + "readonlyRootFilesystem": False, + "privileged": False, + "user": "appuser", + "environment": [], + } ], - ) + )["taskDefinition"]["taskDefinitionArn"] - with mock.patch( - "prowler.providers.aws.services.ecs.ecs_service.ECS", - ecs_client, + 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_containers_readonly_access.ecs_task_definitions_containers_readonly_access.ecs_client", + new=ECS(mocked_aws_provider), ): from prowler.providers.aws.services.ecs.ecs_task_definitions_containers_readonly_access.ecs_task_definitions_containers_readonly_access import ( ecs_task_definitions_containers_readonly_access, @@ -102,38 +125,49 @@ class Test_ecs_task_definitions_containers_readonly_access: result[0].status_extended == f"ECS task definition {TASK_NAME} with revision {TASK_REVISION} has containers with write access to the root filesystem: {CONTAINER_NAME}" ) + assert result[0].resource_id == f"{TASK_NAME}:{TASK_REVISION}" + assert result[0].resource_arn == task_definition_arn + assert result[0].region == AWS_REGION_US_EAST_1 + assert result[0].resource_tags == [] + @mock_aws def test_task_definition_mixed_containers(self): - ecs_client = mock.MagicMock - 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, - readonly_rootfilesystem=False, - privileged=False, - user="appuser", - environment=[], - ), - ContainerDefinition( - name="readonly-container", - readonly_rootfilesystem=True, - privileged=False, - user="appuser", - environment=[], - ), - ], - ) - } + ecs_client = client("ecs", region_name=AWS_REGION_US_EAST_1) - with mock.patch( - "prowler.providers.aws.services.ecs.ecs_service.ECS", - ecs_client, + task_definition_arn = ecs_client.register_task_definition( + family=TASK_NAME, + containerDefinitions=[ + { + "name": CONTAINER_NAME, + "image": "ubuntu", + "memory": 128, + "readonlyRootFilesystem": False, # Not readonly + "privileged": False, + "user": "appuser", + "environment": [], + }, + { + "name": "readonly-container", + "image": "ubuntu", + "memory": 128, + "readonlyRootFilesystem": True, # Readonly + "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_containers_readonly_access.ecs_task_definitions_containers_readonly_access.ecs_client", + new=ECS(mocked_aws_provider), ): from prowler.providers.aws.services.ecs.ecs_task_definitions_containers_readonly_access.ecs_task_definitions_containers_readonly_access import ( ecs_task_definitions_containers_readonly_access, @@ -147,3 +181,7 @@ class Test_ecs_task_definitions_containers_readonly_access: result[0].status_extended == f"ECS task definition {TASK_NAME} with revision {TASK_REVISION} has containers with write access to the root filesystem: {CONTAINER_NAME}" ) + assert result[0].resource_id == f"{TASK_NAME}:{TASK_REVISION}" + assert result[0].resource_arn == task_definition_arn + assert result[0].region == AWS_REGION_US_EAST_1 + assert result[0].resource_tags == [] diff --git a/tests/providers/aws/services/ecs/ecs_task_definitions_host_namespace_not_shared/ecs_task_definitions_host_namespace_not_shared_test.py b/tests/providers/aws/services/ecs/ecs_task_definitions_host_namespace_not_shared/ecs_task_definitions_host_namespace_not_shared_test.py index c88a9a6c7e..92346124e0 100644 --- a/tests/providers/aws/services/ecs/ecs_task_definitions_host_namespace_not_shared/ecs_task_definitions_host_namespace_not_shared_test.py +++ b/tests/providers/aws/services/ecs/ecs_task_definitions_host_namespace_not_shared/ecs_task_definitions_host_namespace_not_shared_test.py @@ -1,33 +1,28 @@ -from unittest import mock +from unittest.mock import patch -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, -) +from boto3 import client +from moto import mock_aws -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}" +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_host_namespace_not_shared: + @mock_aws def test_no_task_definitions(self): - ecs_client = mock.MagicMock - ecs_client.task_definitions = {} + from prowler.providers.aws.services.ecs.ecs_service import ECS - with mock.patch( + mocked_aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1]) + + with 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, + return_value=mocked_aws_provider, + ), patch( + "prowler.providers.aws.services.ecs.ecs_task_definitions_host_namespace_not_shared.ecs_task_definitions_host_namespace_not_shared.ecs_client", + new=ECS(mocked_aws_provider), ): from prowler.providers.aws.services.ecs.ecs_task_definitions_host_namespace_not_shared.ecs_task_definitions_host_namespace_not_shared import ( ecs_task_definitions_host_namespace_not_shared, @@ -37,37 +32,36 @@ class Test_ecs_task_definitions_host_namespace_not_shared: result = check.execute() assert len(result) == 0 + @mock_aws def test_task_definition_no_host_pid_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", - pid_mode="task", - container_definitions=[ - ContainerDefinition( - name=container_name, - privileged=False, - user="", - environment=[ - ContainerEnvVariable( - name="env_var_name_no_secrets", - value="env_var_value_no_secrets", - ) - ], - ) - ], - ) + ecs_client = client("ecs", region_name=AWS_REGION_US_EAST_1) - with mock.patch( + 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": [], + } + ], + pidMode="task", + )["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=set_mocked_aws_provider([AWS_REGION_US_EAST_1]), - ), mock.patch( - "prowler.providers.aws.services.ecs.ecs_service.ECS", - ecs_client, + return_value=mocked_aws_provider, + ), patch( + "prowler.providers.aws.services.ecs.ecs_task_definitions_host_namespace_not_shared.ecs_task_definitions_host_namespace_not_shared.ecs_client", + new=ECS(mocked_aws_provider), ): from prowler.providers.aws.services.ecs.ecs_task_definitions_host_namespace_not_shared.ecs_task_definitions_host_namespace_not_shared import ( ecs_task_definitions_host_namespace_not_shared, @@ -79,35 +73,43 @@ class Test_ecs_task_definitions_host_namespace_not_shared: assert result[0].status == "PASS" assert ( result[0].status_extended - == f"ECS task definition {task_name} with revision {task_revision} does not share a host's process namespace with its containers." + == f"ECS task definition {TASK_NAME} with revision {TASK_REVISION} does not share a host's process namespace with its containers." ) + 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_host_pid_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="host", - pid_mode="host", - container_definitions=[ - ContainerDefinition( - name=container_name, - privileged=False, - user="root", - environment=[], - ) - ], - ) + ecs_client = client("ecs", region_name=AWS_REGION_US_EAST_1) - with mock.patch( + 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": [], + } + ], + pidMode="host", + )["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=set_mocked_aws_provider([AWS_REGION_US_EAST_1]), - ), mock.patch( - "prowler.providers.aws.services.ecs.ecs_service.ECS", - ecs_client, + return_value=mocked_aws_provider, + ), patch( + "prowler.providers.aws.services.ecs.ecs_task_definitions_host_namespace_not_shared.ecs_task_definitions_host_namespace_not_shared.ecs_client", + new=ECS(mocked_aws_provider), ): from prowler.providers.aws.services.ecs.ecs_task_definitions_host_namespace_not_shared.ecs_task_definitions_host_namespace_not_shared import ( ecs_task_definitions_host_namespace_not_shared, @@ -119,35 +121,42 @@ class Test_ecs_task_definitions_host_namespace_not_shared: assert result[0].status == "FAIL" assert ( result[0].status_extended - == f"ECS task definition {task_name} with revision {task_revision} is configured to share a host's process namespace with its containers." + == f"ECS task definition {TASK_NAME} with revision {TASK_REVISION} is configured to share a host's process namespace with its containers." ) + 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_no_pid_mode(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", - pid_mode="", - container_definitions=[ - ContainerDefinition( - name=container_name, - privileged=True, - user="root", - environment=[], - ) - ], - ) + ecs_client = client("ecs", region_name=AWS_REGION_US_EAST_1) - with mock.patch( + 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": [], + } + ], + )["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=set_mocked_aws_provider([AWS_REGION_US_EAST_1]), - ), mock.patch( - "prowler.providers.aws.services.ecs.ecs_service.ECS", - ecs_client, + return_value=mocked_aws_provider, + ), patch( + "prowler.providers.aws.services.ecs.ecs_task_definitions_host_namespace_not_shared.ecs_task_definitions_host_namespace_not_shared.ecs_client", + new=ECS(mocked_aws_provider), ): from prowler.providers.aws.services.ecs.ecs_task_definitions_host_namespace_not_shared.ecs_task_definitions_host_namespace_not_shared import ( ecs_task_definitions_host_namespace_not_shared, @@ -159,5 +168,9 @@ class Test_ecs_task_definitions_host_namespace_not_shared: assert result[0].status == "PASS" assert ( result[0].status_extended - == f"ECS task definition {task_name} with revision {task_revision} does not share a host's process namespace with its containers." + == f"ECS task definition {TASK_NAME} with revision {TASK_REVISION} does not share a host's process namespace with its containers." ) + 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 == [] diff --git a/tests/providers/aws/services/ecs/ecs_task_definitions_host_networking_mode_users/ecs_task_definitions_host_networking_mode_users_test.py b/tests/providers/aws/services/ecs/ecs_task_definitions_host_networking_mode_users/ecs_task_definitions_host_networking_mode_users_test.py index 827d085de3..6458a4e527 100644 --- a/tests/providers/aws/services/ecs/ecs_task_definitions_host_networking_mode_users/ecs_task_definitions_host_networking_mode_users_test.py +++ b/tests/providers/aws/services/ecs/ecs_task_definitions_host_networking_mode_users/ecs_task_definitions_host_networking_mode_users_test.py @@ -1,26 +1,27 @@ -from unittest import mock +from unittest.mock import patch -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 +from boto3 import client +from moto import mock_aws -TASK_NAME = "test-task-hostmode" +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" -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 = {} + from prowler.providers.aws.services.ecs.ecs_service import ECS - with mock.patch( - "prowler.providers.aws.services.ecs.ecs_service.ECS", - ecs_client, + 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_host_networking_mode_users.ecs_task_definitions_host_networking_mode_users.ecs_client", + new=ECS(mocked_aws_provider), ): 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, @@ -30,33 +31,36 @@ class Test_ecs_task_definitions_host_networking_mode_users: result = check.execute() assert len(result) == 0 + @mock_aws 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", - ) - ], - ) - ], - ) + ecs_client = client("ecs", region_name=AWS_REGION_US_EAST_1) - with mock.patch( - "prowler.providers.aws.services.ecs.ecs_service.ECS", - ecs_client, + task_arn = ecs_client.register_task_definition( + family=TASK_NAME, + networkMode="bridge", + 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_host_networking_mode_users.ecs_task_definitions_host_networking_mode_users.ecs_client", + new=ECS(mocked_aws_provider), ): 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, @@ -70,29 +74,41 @@ class Test_ecs_task_definitions_host_networking_mode_users: result[0].status_extended == f"ECS task definition {TASK_NAME} with revision {TASK_REVISION} does not have host network 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 == [] + @mock_aws 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=[], - ) - ], - ) + ecs_client = client("ecs", region_name=AWS_REGION_US_EAST_1) - with mock.patch( - "prowler.providers.aws.services.ecs.ecs_service.ECS", - ecs_client, + task_arn = ecs_client.register_task_definition( + family=TASK_NAME, + networkMode="host", + containerDefinitions=[ + { + "name": CONTAINER_NAME, + "image": "ubuntu", + "memory": 128, + "readonlyRootFilesystem": True, + "privileged": False, + "user": "root", + "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_host_networking_mode_users.ecs_task_definitions_host_networking_mode_users.ecs_client", + new=ECS(mocked_aws_provider), ): 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, @@ -106,29 +122,41 @@ class Test_ecs_task_definitions_host_networking_mode_users: result[0].status_extended == f"ECS task definition {TASK_NAME} with revision {TASK_REVISION} has containers with host network mode and non-privileged containers running as root or with no user specified: {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_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=[], - ) - ], - ) + ecs_client = client("ecs", region_name=AWS_REGION_US_EAST_1) - with mock.patch( - "prowler.providers.aws.services.ecs.ecs_service.ECS", - ecs_client, + task_arn = ecs_client.register_task_definition( + family=TASK_NAME, + networkMode="host", + containerDefinitions=[ + { + "name": CONTAINER_NAME, + "image": "ubuntu", + "memory": 128, + "readonlyRootFilesystem": True, + "privileged": True, + "user": "root", + "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_host_networking_mode_users.ecs_task_definitions_host_networking_mode_users.ecs_client", + new=ECS(mocked_aws_provider), ): 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, @@ -142,29 +170,41 @@ class Test_ecs_task_definitions_host_networking_mode_users: result[0].status_extended == f"ECS task definition {TASK_NAME} with revision {TASK_REVISION} has host network mode but no containers running as root or with no user specified." ) + 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_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=[], - ) - ], - ) + ecs_client = client("ecs", region_name=AWS_REGION_US_EAST_1) - with mock.patch( - "prowler.providers.aws.services.ecs.ecs_service.ECS", - ecs_client, + task_arn = ecs_client.register_task_definition( + family=TASK_NAME, + networkMode="host", + 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_host_networking_mode_users.ecs_task_definitions_host_networking_mode_users.ecs_client", + new=ECS(mocked_aws_provider), ): 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, @@ -178,3 +218,7 @@ class Test_ecs_task_definitions_host_networking_mode_users: result[0].status_extended == f"ECS task definition {TASK_NAME} with revision {TASK_REVISION} has host network mode but no containers running as root or with no user specified." ) + 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 == [] diff --git a/tests/providers/aws/services/ecs/ecs_task_definitions_logging_enabled/ecs_task_definitions_logging_enabled_test.py b/tests/providers/aws/services/ecs/ecs_task_definitions_logging_enabled/ecs_task_definitions_logging_enabled_test.py index 037ac54bd5..ce8891340d 100644 --- a/tests/providers/aws/services/ecs/ecs_task_definitions_logging_enabled/ecs_task_definitions_logging_enabled_test.py +++ b/tests/providers/aws/services/ecs/ecs_task_definitions_logging_enabled/ecs_task_definitions_logging_enabled_test.py @@ -1,21 +1,14 @@ -from unittest import mock +from unittest.mock import patch import botocore +from boto3 import client +from moto import mock_aws -from prowler.providers.aws.services.ecs.ecs_service import ( - ContainerDefinition, - TaskDefinition, -) -from tests.providers.aws.utils import ( - AWS_ACCOUNT_NUMBER, - AWS_REGION_US_EAST_1, - set_mocked_aws_provider, -) +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" -TASK_ARN = f"arn:aws:ecs:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:task-definition/{TASK_NAME}:{TASK_REVISION}" make_api_call = botocore.client.BaseClient._make_api_call @@ -49,12 +42,16 @@ def mock_make_api_call(self, operation_name, kwarg): class Test_ecs_task_definitions_logging_enabled: def test_no_task_definitions(self): - ecs_client = mock.MagicMock - ecs_client.task_definitions = {} + from prowler.providers.aws.services.ecs.ecs_service import ECS - with mock.patch( - "prowler.providers.aws.services.ecs.ecs_service.ECS", - ecs_client, + 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_enabled.ecs_task_definitions_logging_enabled.ecs_client", + new=ECS(mocked_aws_provider), ): from prowler.providers.aws.services.ecs.ecs_task_definitions_logging_enabled.ecs_task_definitions_logging_enabled import ( ecs_task_definitions_logging_enabled, @@ -64,56 +61,35 @@ class Test_ecs_task_definitions_logging_enabled: result = check.execute() assert len(result) == 0 - @mock.patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call) + @mock_aws def test_task_definition_no_logconfiguration(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": [], + } + ], + )["taskDefinition"]["taskDefinitionArn"] from prowler.providers.aws.services.ecs.ecs_service import ECS - aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1]) + mocked_aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1]) - with mock.patch( + with patch( "prowler.providers.common.provider.Provider.get_global_provider", - return_value=aws_provider, - ), mock.patch( + return_value=mocked_aws_provider, + ), patch( "prowler.providers.aws.services.ecs.ecs_task_definitions_logging_enabled.ecs_task_definitions_logging_enabled.ecs_client", - new=ECS(aws_provider), - ): - from prowler.providers.aws.services.ecs.ecs_task_definitions_logging_enabled.ecs_task_definitions_logging_enabled import ( - ecs_task_definitions_logging_enabled, - ) - - check = ecs_task_definitions_logging_enabled() - 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} has containers running with no logging configuration: {CONTAINER_NAME}" - ) - - def test_task_definition_no_logdriver(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=True, - user="root", - environment=[], - log_driver="", - ) - ], - ) - - with mock.patch( - "prowler.providers.aws.services.ecs.ecs_service.ECS", - ecs_client, + new=ECS(mocked_aws_provider), ): from prowler.providers.aws.services.ecs.ecs_task_definitions_logging_enabled.ecs_task_definitions_logging_enabled import ( ecs_task_definitions_logging_enabled, @@ -127,30 +103,41 @@ class Test_ecs_task_definitions_logging_enabled: result[0].status_extended == f"ECS task definition {TASK_NAME} with revision {TASK_REVISION} has containers running with no logging configuration: {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_privileged_container(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=True, - user="root", - environment=[], - log_driver="awslogs", - ) - ], - ) + ecs_client = client("ecs", region_name=AWS_REGION_US_EAST_1) - with mock.patch( - "prowler.providers.aws.services.ecs.ecs_service.ECS", - ecs_client, + 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"}, + } + ], + )["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_enabled.ecs_task_definitions_logging_enabled.ecs_client", + new=ECS(mocked_aws_provider), ): from prowler.providers.aws.services.ecs.ecs_task_definitions_logging_enabled.ecs_task_definitions_logging_enabled import ( ecs_task_definitions_logging_enabled, @@ -164,3 +151,7 @@ class Test_ecs_task_definitions_logging_enabled: result[0].status_extended == f"ECS task definition {TASK_NAME} with revision {TASK_REVISION} containers have logging configured." ) + 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 == [] diff --git a/tests/providers/aws/services/ecs/ecs_task_definitions_no_environment_secrets/ecs_task_definitions_no_environment_secrets_test.py b/tests/providers/aws/services/ecs/ecs_task_definitions_no_environment_secrets/ecs_task_definitions_no_environment_secrets_test.py index 7defe40eac..8f8490a726 100644 --- a/tests/providers/aws/services/ecs/ecs_task_definitions_no_environment_secrets/ecs_task_definitions_no_environment_secrets_test.py +++ b/tests/providers/aws/services/ecs/ecs_task_definitions_no_environment_secrets/ecs_task_definitions_no_environment_secrets_test.py @@ -1,30 +1,31 @@ -from unittest import mock +from unittest.mock import patch -from prowler.providers.aws.services.ecs.ecs_service import ( - ContainerDefinition, - ContainerEnvVariable, - TaskDefinition, -) +from boto3 import client +from moto import mock_aws -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" -env_var_value_with_secrets = "pass-12343" +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" +ENV_VAR_NAME_NO_SECRETS = "host" +ENV_VAR_VALUE_NO_SECRETS = "localhost:1234" +ENV_VAR_NAME_WITH_SECRETS = "DB_PASSWORD" +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 = {} + from prowler.providers.aws.services.ecs.ecs_service import ECS - with mock.patch( - "prowler.providers.aws.services.ecs.ecs_service.ECS", - ecs_client, + 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_no_environment_secrets.ecs_task_definitions_no_environment_secrets.ecs_client", + new=ECS(mocked_aws_provider), ): from prowler.providers.aws.services.ecs.ecs_task_definitions_no_environment_secrets.ecs_task_definitions_no_environment_secrets import ( ecs_task_definitions_no_environment_secrets, @@ -34,31 +35,40 @@ class Test_ecs_task_definitions_no_environment_secrets: result = check.execute() assert len(result) == 0 + @mock_aws def test_container_env_var_no_secrets(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}:{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 - ) - ], - ) - ], - ) + ecs_client = client("ecs", region_name=AWS_REGION_US_EAST_1) - with mock.patch( - "prowler.providers.aws.services.ecs.ecs_service.ECS", - ecs_client, + 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": [ + { + "name": ENV_VAR_NAME_NO_SECRETS, + "value": ENV_VAR_VALUE_NO_SECRETS, + } + ], + } + ], + )["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_no_environment_secrets.ecs_task_definitions_no_environment_secrets.ecs_client", + new=ECS(mocked_aws_provider), ): from prowler.providers.aws.services.ecs.ecs_task_definitions_no_environment_secrets.ecs_task_definitions_no_environment_secrets import ( ecs_task_definitions_no_environment_secrets, @@ -70,40 +80,47 @@ class Test_ecs_task_definitions_no_environment_secrets: assert result[0].status == "PASS" assert ( result[0].status_extended - == f"No secrets found in variables of ECS task definition {task_name} with revision {task_revision}." - ) - assert result[0].resource_id == f"{task_name}:1" - assert ( - result[0].resource_arn - == f"arn:aws:ecs:{AWS_REGION}:{AWS_ACCOUNT_NUMBER}:task-definition/{task_name}:{task_revision}" + == f"No secrets found in variables of ECS task definition {TASK_NAME} with revision {TASK_REVISION}." ) + 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_container_env_var_with_secrets(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}:{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, - ) - ], - ) - ], - ) + ecs_client = client("ecs", region_name=AWS_REGION_US_EAST_1) - with mock.patch( - "prowler.providers.aws.services.ecs.ecs_service.ECS", - ecs_client, + 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": [ + { + "name": ENV_VAR_NAME_WITH_SECRETS, + "value": ENV_VAR_VALUE_WITH_SECRETS, + } + ], + } + ], + )["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_no_environment_secrets.ecs_task_definitions_no_environment_secrets.ecs_client", + new=ECS(mocked_aws_provider), ): from prowler.providers.aws.services.ecs.ecs_task_definitions_no_environment_secrets.ecs_task_definitions_no_environment_secrets import ( ecs_task_definitions_no_environment_secrets, @@ -115,10 +132,9 @@ class Test_ecs_task_definitions_no_environment_secrets: assert result[0].status == "FAIL" assert ( result[0].status_extended - == 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 ( - result[0].resource_arn - == f"arn:aws:ecs:{AWS_REGION}:{AWS_ACCOUNT_NUMBER}:task-definition/{task_name}:{task_revision}" + == f"Potential secrets found in ECS task definition {TASK_NAME} with revision {TASK_REVISION}: Secrets in container test-container -> Secret Keyword on line 2." ) + 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 == [] diff --git a/tests/providers/aws/services/ecs/ecs_task_definitions_no_privileged_containers/ecs_task_definitions_no_privileged_containers_test.py b/tests/providers/aws/services/ecs/ecs_task_definitions_no_privileged_containers/ecs_task_definitions_no_privileged_containers_test.py index e167283ef8..a18f0c1205 100644 --- a/tests/providers/aws/services/ecs/ecs_task_definitions_no_privileged_containers/ecs_task_definitions_no_privileged_containers_test.py +++ b/tests/providers/aws/services/ecs/ecs_task_definitions_no_privileged_containers/ecs_task_definitions_no_privileged_containers_test.py @@ -1,26 +1,27 @@ -from unittest import mock +from unittest.mock import patch -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 +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" -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_no_privileged_containers: def test_no_task_definitions(self): - ecs_client = mock.MagicMock - ecs_client.task_definitions = {} + from prowler.providers.aws.services.ecs.ecs_service import ECS - with mock.patch( - "prowler.providers.aws.services.ecs.ecs_service.ECS", - ecs_client, + 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_no_privileged_containers.ecs_task_definitions_no_privileged_containers.ecs_client", + new=ECS(mocked_aws_provider), ): from prowler.providers.aws.services.ecs.ecs_task_definitions_no_privileged_containers.ecs_task_definitions_no_privileged_containers import ( ecs_task_definitions_no_privileged_containers, @@ -30,33 +31,35 @@ class Test_ecs_task_definitions_no_privileged_containers: result = check.execute() assert len(result) == 0 + @mock_aws def test_task_definition_no_priviled_container(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", - ) - ], - ) - ], - ) + ecs_client = client("ecs", region_name=AWS_REGION_US_EAST_1) - with mock.patch( - "prowler.providers.aws.services.ecs.ecs_service.ECS", - ecs_client, + 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": [], + } + ], + )["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_no_privileged_containers.ecs_task_definitions_no_privileged_containers.ecs_client", + new=ECS(mocked_aws_provider), ): from prowler.providers.aws.services.ecs.ecs_task_definitions_no_privileged_containers.ecs_task_definitions_no_privileged_containers import ( ecs_task_definitions_no_privileged_containers, @@ -70,29 +73,40 @@ class Test_ecs_task_definitions_no_privileged_containers: result[0].status_extended == f"ECS task definition {TASK_NAME} with revision {TASK_REVISION} does not have privileged containers." ) + assert result[0].resource_id == f"{TASK_NAME}:{TASK_REVISION}" + assert result[0].resource_arn == task_arn + assert result[0].resource_tags == [] + assert result[0].region == AWS_REGION_US_EAST_1 + @mock_aws def test_task_definition_privileged_container(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=True, - user="root", - environment=[], - ) - ], - ) + ecs_client = client("ecs", region_name=AWS_REGION_US_EAST_1) - with mock.patch( - "prowler.providers.aws.services.ecs.ecs_service.ECS", - ecs_client, + 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": [], + } + ], + )["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_no_privileged_containers.ecs_task_definitions_no_privileged_containers.ecs_client", + new=ECS(mocked_aws_provider), ): from prowler.providers.aws.services.ecs.ecs_task_definitions_no_privileged_containers.ecs_task_definitions_no_privileged_containers import ( ecs_task_definitions_no_privileged_containers, @@ -106,3 +120,7 @@ class Test_ecs_task_definitions_no_privileged_containers: result[0].status_extended == f"ECS task definition {TASK_NAME} with revision {TASK_REVISION} has privileged containers: {CONTAINER_NAME}" ) + assert result[0].resource_id == f"{TASK_NAME}:{TASK_REVISION}" + assert result[0].resource_arn == task_arn + assert result[0].resource_tags == [] + assert result[0].region == AWS_REGION_US_EAST_1