Vulnerability scanning AWS ECS running Images

Get the fixable vulnerabilities of you running ECS tasks

published: Sat, 31 Aug 2024 estimated reading time: 3 minutes

When you have quite a few applications running on AWS ECS you should want to fix possible vulnerabilities that your docker images have. There are quite a few options to get the vulnerabilities. We will not look into all the options that are there in this post. What we do look at is getting only the vulnerabilities for running images. If you scan your docker repository it might contain images that you don’t run anymore but still have in the registry. We will also only look at only vulnerabilities that we can fix, because if you won’t fix it in the packages yourself you are unable to do anything with the vulnerabilities anyways.

Prerequisite

  • Install Trivy
  • Have a working AWS CLI, Python and boto3 installed

create a python file for the script we are going to create with the following code:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
import boto3
from collections import defaultdict
import subprocess
import json

client = boto3.client('ecs')

def main():
    pass

if __name__ == '__main__':
    main()

List ECS clusters

First we need to get all the running tasks so we can later get all the task running on the clusters.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
def get_ecs_cluster_arns():
    paginator = client.get_paginator('list_clusters')
    cluster_response_iterator = paginator.paginate(
        PaginationConfig={
            'PageSize': 30,
        }
    )
    cluster_arns = []
    for page in cluster_response_iterator:
        for arn in page["clusterArns"]:
            cluster_arns.append(arn)
    return cluster_arns

Update the main with

cluster_arns = get_ecs_cluster_arns()

List the running tasks in a cluster

After that we can get all the tasks that run in a cluster and do this for all the clusters.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
def get_task_arns(cluster_arn):
    paginator = client.get_paginator('list_tasks')
    task_response_iterator = paginator.paginate(
        cluster=cluster_arn,
        PaginationConfig={
            'PageSize': 30,
        }
    )
    task_arns = defaultdict(list)

    for page in task_response_iterator:
        for arn in page["taskArns"]:
            task_arns[cluster_arn].append(arn)
    return task_arns

Update the main with

cluster_task_arns = {}
for arn in cluster_arns:
    cluster_task_arns.update( get_task_arns(arn))

Get all the docker images run for the tasks

For all the tasks can we now get the docker images. Since some can be duplicate we can use a set to prevent duplicates.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
def get_docker_images(cluster_task_arns):
    images = set()
    for key, value in cluster_task_arns.items():
        response = client.describe_tasks(
            cluster=key,
            tasks=value
        )
        for task in response['tasks']:
            for container in task['containers']:
                images.add(container["image"])

    return images

Update the main with

docker_images = get_docker_images(cluster_task_arns)

Remove images you don’t want to scan

Maybe there are some images you don’t want to scan. YOu can remove the before starting the scanning.

1
2
3
4
5
6
7
def remove_images_to_not_scan(docker_images):
    not_scan__images = ["aws-guardduty-agent"]
    img_copy = list(docker_images)
    for docker_image in img_copy:
        for part_image_name in not_scan__images:
            if part_image_name in docker_image:
                docker_images.remove(docker_image)

Update the main with

remove_images_to_not_scan(docker_images)

Scan the images

Finally we scan the docker images. Since Trivy give a lot of information back we keep only the lines we want to keep.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
def scan_docker_images(docker_images):
    docker_image_vulnerabilities = {}
    keep_for_vulnerability_overview = ['VulnerabilityID', 'PkgName', 'Severity', 'InstalledVersion', 'FixedVersion', 'DataSource']
    for docker_image in docker_images:
        result = subprocess.run(['trivy', 'image', docker_image, "--ignore-unfixed", "-f", "json"], capture_output=True,
                                text=True)
        result_output = json.loads(result.stdout)
        trivy_scan_result = result_output["Results"]

        vulnerabilities = {}
        for package_type in trivy_scan_result:
            if package_type.get("Vulnerabilities"):
                to_keep = []
                for vuln in package_type["Vulnerabilities"]:
                    obj = {}
                    for key, value in vuln.items():
                        if key in keep_for_vulnerability_overview:
                            obj[key] = value
                    to_keep.append(obj)
                vulnerabilities[package_type["Type"]] = to_keep
        docker_image_vulnerabilities[docker_image] = vulnerabilities
    return docker_image_vulnerabilities

Update the main with

docker_image_vulnerabilities = scan_docker_images(docker_images)

Filter the images that have no fixable vulnerabilities

If there is nothing to fix we don’t want to see the images in our list.

1
2
3
4
5
6
def remove_not_vulnerable_and_fixable_images(docker_image_vulnerabilities):
    only_vulnerable_images = {}
    for key, value in docker_image_vulnerabilities.items():
        if value:
            only_vulnerable_images[key] = value
    return only_vulnerable_images

Update the main with

only_vulnerable_images=remove_not_vulnerable_and_fixable_images(docker_image_vulnerabilities)

Write the output to a files

To keep our results for whatever we want to do with it we write the output to a json file. Add this to the main function to complete the script.

with open('my_vulnerabilities.json', 'w') as outfile:
    json.dump(only_vulnerable_images, outfile, )

Run this script on any AWS account you want to get the vulnerabilities that you want to fix.