feat(ecs): add new check ecs_task_set_no_assign_public_ip (#5603)

Co-authored-by: Sergio <sergio@prowler.com>
This commit is contained in:
Mario Rodriguez Lopez
2024-11-04 17:49:22 +01:00
committed by GitHub
parent 84d4e4a604
commit bc89f4383e
6 changed files with 339 additions and 0 deletions
@@ -15,6 +15,7 @@ class ECS(AWSService):
self.task_definitions = {}
self.services = {}
self.clusters = {}
self.task_sets = {}
self.__threading_call__(self._list_task_definitions)
self.__threading_call__(
self._describe_task_definition, self.task_definitions.values()
@@ -127,6 +128,18 @@ class ECS(AWSService):
platform_family=service_desc.get("platformFamily", ""),
tags=service_desc.get("tags", []),
)
for task_set in service_desc.get("taskSets", []):
self.task_sets[task_set["taskSetArn"]] = TaskSet(
id=task_set["id"],
arn=task_set["taskSetArn"],
cluster_arn=task_set["clusterArn"],
service_arn=task_set["serviceArn"],
assign_public_ip=task_set.get("networkConfiguration", {})
.get("awsvpcConfiguration", {})
.get("assignPublicIp", "DISABLED"),
region=cluster.region,
tags=task_set.get("tags", []),
)
cluster.services[service_arn] = service_obj
self.services[service_arn] = service_obj
except Exception as error:
@@ -215,3 +228,13 @@ class Cluster(BaseModel):
services: dict = {}
settings: Optional[list] = []
tags: Optional[list] = []
class TaskSet(BaseModel):
id: str
arn: str
cluster_arn: str
service_arn: str
region: str
assign_public_ip: Optional[str]
tags: Optional[list] = []
@@ -0,0 +1,34 @@
{
"Provider": "aws",
"CheckID": "ecs_task_set_no_assign_public_ip",
"CheckTitle": "ECS task sets should not automatically assign public IP addresses",
"CheckType": [
"Software and Configuration Checks/AWS Security Best Practices"
],
"ServiceName": "ecs",
"SubServiceName": "",
"ResourceIdTemplate": "arn:aws:ecs:{region}:{account-id}:task-set/{cluster-name}/{service-name}/{task-set-id}",
"Severity": "high",
"ResourceType": "AwsEcsTaskSet",
"Description": "This control checks whether an Amazon ECS task set is configured to automatically assign public IP addresses. The control fails if AssignPublicIP is set to ENABLED.",
"Risk": "A public IP address is reachable from the internet, potentially exposing resources associated with the ECS task set. ECS task sets shouldn't be publicly accessible, as this may allow unintended access to container application servers.",
"RelatedUrl": "https://docs.aws.amazon.com/AmazonECS/latest/APIReference/API_TaskSet.html",
"Remediation": {
"Code": {
"CLI": "aws ecs update-service --cluster <cluster-name> --service <service-name> --network-configuration 'awsvpcConfiguration={assignPublicIp=\"DISABLED\"}'",
"NativeIaC": "",
"Other": "https://docs.aws.amazon.com/securityhub/latest/userguide/ecs-controls.html#ecs-16",
"Terraform": ""
},
"Recommendation": {
"Text": "Configure ECS task sets to not assign public IP addresses to prevent unintended public access to your containerized applications.",
"Url": "https://docs.aws.amazon.com/AmazonECS/latest/developerguide/update-task-definition-console-v2.html"
}
},
"Categories": [
"internet-exposed"
],
"DependsOn": [],
"RelatedTo": [],
"Notes": ""
}
@@ -0,0 +1,24 @@
from prowler.lib.check.models import Check, Check_Report_AWS
from prowler.providers.aws.services.ecs.ecs_client import ecs_client
class ecs_task_set_no_assign_public_ip(Check):
def execute(self):
findings = []
for task_set in ecs_client.task_sets.values():
report = Check_Report_AWS(self.metadata())
report.region = task_set.region
report.resource_id = task_set.id
report.resource_arn = task_set.arn
report.resource_tags = task_set.tags
report.status = "PASS"
report.status_extended = f"ECS Task Set {task_set.id} does not have automatic public IP assignment."
if task_set.assign_public_ip == "ENABLED":
report.status = "FAIL"
report.status_extended = (
f"ECS Task Set {task_set.id} has automatic public IP assignment."
)
findings.append(report)
return findings
@@ -61,6 +61,22 @@ def mock_make_api_call(self, operation_name, kwarg):
"launchType": "FARGATE",
"platformVersion": "1.4.0",
"platformFamily": "Linux",
"taskSets": [
{
"id": "ecs-svc/task-set",
"taskSetArn": "arn:aws:ecs:eu-west-1:123456789012:task-set/test_cluster_1/test_ecs_service/ecs-svc/task-set",
"clusterArn": "arn:aws:ecs:eu-west-1:123456789012:cluster/test_cluster_1",
"serviceArn": "arn:aws:ecs:eu-west-1:123456789012:service/test_cluster_1/test_ecs_service",
"networkConfiguration": {
"awsvpcConfiguration": {
"subnets": ["subnet-12345678"],
"securityGroups": ["sg-12345678"],
"assignPublicIp": "DISABLED",
},
},
"tags": [],
}
],
}
]
}
@@ -228,6 +244,8 @@ class Test_ECS_Service:
"arn:aws:ecs:eu-west-1:123456789012:service/test_cluster_1/test_ecs_service"
)
task_set_arn = "arn:aws:ecs:eu-west-1:123456789012:task-set/test_cluster_1/test_ecs_service/ecs-svc/task-set"
assert len(ecs.services) == 1
assert ecs.services[service_arn].name == "test_ecs_service"
assert ecs.services[service_arn].arn == service_arn
@@ -237,3 +255,14 @@ class Test_ECS_Service:
assert ecs.services[service_arn].launch_type == "FARGATE"
assert ecs.services[service_arn].platform_version == "1.4.0"
assert ecs.services[service_arn].platform_family == "Linux"
assert len(ecs.task_sets) == 1
assert ecs.task_sets[task_set_arn].id == "ecs-svc/task-set"
assert ecs.task_sets[task_set_arn].arn == task_set_arn
assert (
ecs.task_sets[task_set_arn].cluster_arn
== "arn:aws:ecs:eu-west-1:123456789012:cluster/test_cluster_1"
)
assert ecs.task_sets[task_set_arn].service_arn == service_arn
assert ecs.task_sets[task_set_arn].assign_public_ip == "DISABLED"
assert ecs.task_sets[task_set_arn].region == AWS_REGION_EU_WEST_1
assert ecs.task_sets[task_set_arn].tags == []
@@ -0,0 +1,229 @@
from unittest.mock import patch
import botocore
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,
)
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-task-set-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-task-set-no-public-ip",
"desiredCount": 1,
"launchType": "FARGATE",
"platformVersion": "1.4.0",
"platformFamily": "Linux",
"networkConfiguration": {
"awsvpcConfiguration": {
"subnets": ["subnet-12345678"],
"securityGroups": ["sg-12345678"],
"assignPublicIp": "DISABLED",
},
},
"taskSets": [
{
"id": "ecs-svc/task-set-no-public-ip",
"taskSetArn": f"arn:aws:ecs:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:task-set/sample-cluster/service-task-set-no-public-ip/ecs-svc/task-set-no-public-ip",
"clusterArn": f"arn:aws:ecs:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:cluster/sample-cluster",
"serviceArn": f"arn:aws:ecs:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:service/sample-cluster/service-task-set-no-public-ip",
"networkConfiguration": {
"awsvpcConfiguration": {
"subnets": ["subnet-12345678"],
"securityGroups": ["sg-12345678"],
"assignPublicIp": "DISABLED",
},
},
"tags": [],
}
],
"tags": [],
},
],
}
elif kwarg["services"] == [
f"arn:aws:ecs:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:service/sample-cluster/service-task-set-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",
},
},
"taskSets": [
{
"id": "ecs-svc/task-set-public-ip",
"taskSetArn": f"arn:aws:ecs:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:task-set/sample-cluster/service-task-set-public-ip/ecs-svc/task-set-public-ip",
"clusterArn": f"arn:aws:ecs:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:cluster/sample-cluster",
"serviceArn": f"arn:aws:ecs:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:service/sample-cluster/service-task-set-public-ip",
"networkConfiguration": {
"awsvpcConfiguration": {
"subnets": ["subnet-12345678"],
"securityGroups": ["sg-12345678"],
"assignPublicIp": "ENABLED",
},
},
"tags": [],
}
],
"tags": [],
},
],
}
return orig(self, operation_name, kwarg)
class Test_ecs_task_sets_no_assign_public_ip:
@mock_aws
def test_no_services(self):
from prowler.providers.aws.services.ecs.ecs_service import ECS
mocked_aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
with patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mocked_aws_provider,
), patch(
"prowler.providers.aws.services.ecs.ecs_task_set_no_assign_public_ip.ecs_task_set_no_assign_public_ip.ecs_client",
new=ECS(mocked_aws_provider),
):
from prowler.providers.aws.services.ecs.ecs_task_set_no_assign_public_ip.ecs_task_set_no_assign_public_ip import (
ecs_task_set_no_assign_public_ip,
)
check = ecs_task_set_no_assign_public_ip()
result = check.execute()
assert len(result) == 0
@mock_aws
@patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call)
def test_task_set_with_no_public_ip(self):
ecs_client = client("ecs", region_name=AWS_REGION_US_EAST_1)
ecs_client.create_cluster(clusterName="sample-cluster")
ecs_client.create_service(
cluster="sample-cluster",
serviceName="service-task-set-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_task_set_no_assign_public_ip.ecs_task_set_no_assign_public_ip.ecs_client",
new=ECS(mocked_aws_provider),
):
from prowler.providers.aws.services.ecs.ecs_task_set_no_assign_public_ip.ecs_task_set_no_assign_public_ip import (
ecs_task_set_no_assign_public_ip,
)
check = ecs_task_set_no_assign_public_ip()
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert (
result[0].status_extended
== "ECS Task Set ecs-svc/task-set-no-public-ip does not have automatic public IP assignment."
)
assert result[0].resource_id == "ecs-svc/task-set-no-public-ip"
assert (
result[0].resource_arn
== f"arn:aws:ecs:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:task-set/sample-cluster/service-task-set-no-public-ip/ecs-svc/task-set-no-public-ip"
)
assert result[0].resource_tags == []
assert result[0].region == AWS_REGION_US_EAST_1
@mock_aws
@patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call)
def test_task_set_public_ip(self):
ecs_client = client("ecs", region_name=AWS_REGION_US_EAST_1)
ecs_client.create_cluster(clusterName="sample-cluster")
ecs_client.create_service(
cluster="sample-cluster",
serviceName="service-task-set-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_task_set_no_assign_public_ip.ecs_task_set_no_assign_public_ip.ecs_client",
new=ECS(mocked_aws_provider),
):
from prowler.providers.aws.services.ecs.ecs_task_set_no_assign_public_ip.ecs_task_set_no_assign_public_ip import (
ecs_task_set_no_assign_public_ip,
)
check = ecs_task_set_no_assign_public_ip()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== "ECS Task Set ecs-svc/task-set-public-ip has automatic public IP assignment."
)
assert result[0].resource_id == "ecs-svc/task-set-public-ip"
assert (
result[0].resource_arn
== f"arn:aws:ecs:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:task-set/sample-cluster/service-task-set-public-ip/ecs-svc/task-set-public-ip"
)
assert result[0].resource_tags == []
assert result[0].region == AWS_REGION_US_EAST_1