ben.carrasco/prowler
1
0
Fork 0
mirror of https://github.com/prowler-cloud/prowler.git synced 2026-07-24 13:01:56 +00:00
Code Issues Packages Projects Releases Wiki Activity

chore(ecs): mock all tests using moto (#5333)

Browse Source 
Co-authored-by: Rubén De la Torre Vico <rubendltv22@gmail.com>
Co-authored-by: Sergio <sergio@prowler.com>
This commit is contained in:
Prowler Bot
2024-10-08 22:34:53 +02:00
committed by GitHub
parent 00232ca7ea
commit 807f2f9f30
8 changed files with 759 additions and 537 deletions
Download Patch File Download Diff File Expand all files Collapse all files
prowler/providers/aws/services/ecs/ecs_service.py
-1
View File
@@ -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__
tests/providers/aws/services/ecs/ecs_service_no_assign_public_ip/ecs_service_no_assign_public_ip_test.py
+144 -41
View File
@@ -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
tests/providers/aws/services/ecs/ecs_task_definitions_containers_readonly_access/ecs_task_definitions_containers_readonly_access_test.py
+118 -80
View File
@@ -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 == []
tests/providers/aws/services/ecs/ecs_task_definitions_host_namespace_not_shared/ecs_task_definitions_host_namespace_not_shared_test.py
+112 -99
View File
@@ -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 == []
tests/providers/aws/services/ecs/ecs_task_definitions_host_networking_mode_users/ecs_task_definitions_host_networking_mode_users_test.py
+143 -99
View File
@@ -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 == []
tests/providers/aws/services/ecs/ecs_task_definitions_logging_enabled/ecs_task_definitions_logging_enabled_test.py
+71 -80
View File
@@ -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 == []
tests/providers/aws/services/ecs/ecs_task_definitions_no_environment_secrets/ecs_task_definitions_no_environment_secrets_test.py
+95 -79
View File
@@ -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 == []
tests/providers/aws/services/ecs/ecs_task_definitions_no_privileged_containers/ecs_task_definitions_no_privileged_containers_test.py
+76 -58
View File
@@ -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