feat(ecs): Ensure ECS clusters use Container Insights (#5241)

Co-authored-by: Sergio <sergio@prowler.com>
This commit is contained in:
Mario Rodriguez Lopez
2024-10-02 16:42:52 +02:00
committed by GitHub
parent 2c2dd82d0c
commit 7f97b0a57f
8 changed files with 208 additions and 2 deletions
@@ -0,0 +1,34 @@
{
"Provider": "aws",
"CheckID": "ecs_cluster_container_insights_enabled",
"CheckTitle": "ECS clusters should use Container Insights",
"CheckType": [
"Software and Configuration Checks/AWS Security Best Practices"
],
"ServiceName": "ecs",
"SubServiceName": "",
"ResourceIdTemplate": "arn:aws:ecs:{region}:{account-id}:cluster/{cluster-name}",
"Severity": "medium",
"ResourceType": "AwsEcsCluster",
"Description": "This control checks if ECS clusters use Container Insights. This control fails if Container Insights are not set up for a cluster.",
"Risk": "Without Container Insights, important performance metrics and diagnostic information from containerized applications may not be captured, which can hinder monitoring and troubleshooting.",
"RelatedUrl": "https://docs.aws.amazon.com/config/latest/developerguide/ecs-container-insights-enabled.html",
"Remediation": {
"Code": {
"CLI": "aws ecs update-cluster-settings --cluster <cluster-name> --settings name=containerInsights,value=enabled",
"NativeIaC": "",
"Other": "https://docs.aws.amazon.com/securityhub/latest/userguide/ecs-controls.html#ecs-12",
"Terraform": ""
},
"Recommendation": {
"Text": "Enable Container Insights for your ECS clusters to collect and monitor key performance metrics and diagnostic data from your containers.",
"Url": "https://docs.aws.amazon.com/AmazonCloudWatch/latest/monitoring/deploy-container-insights-ECS-cluster.html"
}
},
"Categories": [
"logging"
],
"DependsOn": [],
"RelatedTo": [],
"Notes": ""
}
@@ -0,0 +1,27 @@
from prowler.lib.check.models import Check, Check_Report_AWS
from prowler.providers.aws.services.ecs.ecs_client import ecs_client
class ecs_cluster_container_insights_enabled(Check):
def execute(self):
findings = []
for cluster in ecs_client.clusters.values():
report = Check_Report_AWS(self.metadata())
report.region = cluster.region
report.resource_id = cluster.name
report.resource_arn = cluster.arn
report.resource_tags = cluster.tags
report.status = "FAIL"
report.status_extended = (
f"ECS cluster {cluster.name} does not have container insights enabled."
)
if cluster.settings:
for setting in cluster.settings:
if (
setting["name"] == "containerInsights"
and setting["value"] == "enabled"
):
report.status = "PASS"
report.status_extended = f"ECS cluster {cluster.name} has container insights enabled."
findings.append(report)
return findings
@@ -158,6 +158,7 @@ class ECS(AWSService):
"TAGS",
],
)
cluster.settings = response["clusters"][0].get("settings", [])
cluster.tags = response["clusters"][0].get("tags", [])
except Exception as error:
logger.error(
@@ -203,4 +204,5 @@ class Cluster(BaseModel):
arn: str
region: str
services: dict = {}
settings: Optional[list] = []
tags: Optional[list] = []
@@ -0,0 +1,137 @@
from unittest import mock
from boto3 import client
from moto import mock_aws
from tests.providers.aws.utils import AWS_REGION_US_EAST_1, set_mocked_aws_provider
CLUSTER_NAME = "test-cluster"
class Test_ecs_clusters_container_insights_enabled:
@mock_aws
def test_no_clusters(self):
from prowler.providers.aws.services.ecs.ecs_service import ECS
aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=aws_provider,
), mock.patch(
"prowler.providers.aws.services.ecs.ecs_cluster_container_insights_enabled.ecs_cluster_container_insights_enabled.ecs_client",
new=ECS(aws_provider),
):
from prowler.providers.aws.services.ecs.ecs_cluster_container_insights_enabled.ecs_cluster_container_insights_enabled import (
ecs_cluster_container_insights_enabled,
)
check = ecs_cluster_container_insights_enabled()
result = check.execute()
assert len(result) == 0
@mock_aws
def test_cluster_no_settings(self):
ecs_client = client("ecs", region_name=AWS_REGION_US_EAST_1)
cluster_arn = ecs_client.create_cluster(
clusterName=CLUSTER_NAME,
)[
"cluster"
]["clusterArn"]
from prowler.providers.aws.services.ecs.ecs_service import ECS
aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=aws_provider,
), mock.patch(
"prowler.providers.aws.services.ecs.ecs_cluster_container_insights_enabled.ecs_cluster_container_insights_enabled.ecs_client",
new=ECS(aws_provider),
):
from prowler.providers.aws.services.ecs.ecs_cluster_container_insights_enabled.ecs_cluster_container_insights_enabled import (
ecs_cluster_container_insights_enabled,
)
check = ecs_cluster_container_insights_enabled()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert result[0].resource_arn == cluster_arn
assert (
result[0].status_extended
== f"ECS cluster {CLUSTER_NAME} does not have container insights enabled."
)
@mock_aws
def test_cluster_enabled_container_insights(self):
ecs_client = client("ecs", region_name=AWS_REGION_US_EAST_1)
cluster_settings = [
{"name": "containerInsights", "value": "enabled"},
]
cluster_arn = ecs_client.create_cluster(
clusterName=CLUSTER_NAME,
settings=cluster_settings,
)["cluster"]["clusterArn"]
from prowler.providers.aws.services.ecs.ecs_service import ECS
aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=aws_provider,
), mock.patch(
"prowler.providers.aws.services.ecs.ecs_cluster_container_insights_enabled.ecs_cluster_container_insights_enabled.ecs_client",
new=ECS(aws_provider),
):
from prowler.providers.aws.services.ecs.ecs_cluster_container_insights_enabled.ecs_cluster_container_insights_enabled import (
ecs_cluster_container_insights_enabled,
)
check = ecs_cluster_container_insights_enabled()
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert result[0].resource_arn == cluster_arn
assert (
result[0].status_extended
== f"ECS cluster {CLUSTER_NAME} has container insights enabled."
)
@mock_aws
def test_cluster_disabled_container_insights(self):
ecs_client = client("ecs", region_name=AWS_REGION_US_EAST_1)
cluster_settings = [
{"name": "containerInsights", "value": "disabled"},
]
cluster_arn = ecs_client.create_cluster(
clusterName=CLUSTER_NAME,
settings=cluster_settings,
)["cluster"]["clusterArn"]
from prowler.providers.aws.services.ecs.ecs_service import ECS
aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=aws_provider,
), mock.patch(
"prowler.providers.aws.services.ecs.ecs_cluster_container_insights_enabled.ecs_cluster_container_insights_enabled.ecs_client",
new=ECS(aws_provider),
):
from prowler.providers.aws.services.ecs.ecs_cluster_container_insights_enabled.ecs_cluster_container_insights_enabled import (
ecs_cluster_container_insights_enabled,
)
check = ecs_cluster_container_insights_enabled()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert result[0].resource_arn == cluster_arn
assert (
result[0].status_extended
== f"ECS cluster {CLUSTER_NAME} does not have container insights enabled."
)
@@ -69,6 +69,9 @@ def mock_make_api_call(self, operation_name, kwarg):
"clusterName": "test_cluster_1",
"status": "ACTIVE",
"tags": [{"key": "Name", "value": "test_cluster_1"}],
"settings": [
{"name": "containerInsights", "value": "enabled"},
],
"registeredContainerInstancesCount": 5,
"runningTasksCount": 10,
"pendingTasksCount": 1,
@@ -195,6 +198,9 @@ class Test_ECS_Service:
assert ecs.clusters[cluster_arn1].tags == [
{"key": "Name", "value": "test_cluster_1"}
]
assert ecs.clusters[cluster_arn1].settings == [
{"name": "containerInsights", "value": "enabled"}
]
@patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call)
# Test describe ECS services
@@ -27,7 +27,7 @@ def mock_aks_get_clusters(_):
"prowler.providers.azure.services.aks.aks_service.AKS._get_clusters",
new=mock_aks_get_clusters,
)
class Test_AppInsights_Service:
class Test_AKS_Service:
def test_get_client(self):
aks = AKS(set_mocked_azure_provider())
assert (
@@ -62,7 +62,7 @@ def mock_vm_get_disks(_):
"prowler.providers.azure.services.vm.vm_service.VirtualMachines._get_disks",
new=mock_vm_get_disks,
)
class Test_AppInsights_Service:
class Test_VirtualMachines_Service:
def test_get_client(self):
app_insights = VirtualMachines(set_mocked_azure_provider())
assert (