Extract secrets from containers in ecs

envecs.py

#!/usr/bin/env -S uv -q run --script 

# /// script
# dependencies = [
#   "boto3",
#   "pzp",
# ]
# ///

import boto3
from pzp import pzp
from sys import stderr

ecs_client = boto3.client("ecs")
ssm_client = boto3.client("ssm")
secrets_client = boto3.client("secretsmanager")


def get_resolved_secret(secret_arn):
    """Retrieve the secret value from either AWS Secrets Manager or SSM Parameter Store."""
    try:
        if secret_arn.startswith("arn:aws:secretsmanager:"):
            # Secret is from AWS Secrets Manager
            response = secrets_client.get_secret_value(SecretId=secret_arn)
            return response["SecretString"]
        elif secret_arn.startswith("arn:aws:ssm:"):
            # Secret is from SSM Parameter Store
            parameter_name = (
                "/" + secret_arn.split(":parameter/")[-1]
            )  # Extract parameter name
            response = ssm_client.get_parameter(
                Name=parameter_name, WithDecryption=True
            )
            return response["Parameter"]["Value"]
        else:
            raise ValueError(f"Unsupported secret source for ARN: {secret_arn}")
    except Exception as e:
        return f"Error retrieving secret: {e}"


def print_container_details(container):
    if "environment" in container:
        for env_var in container["environment"]:
            print(f"{env_var['name']}={env_var['value']}")

    # Print resolved secrets
    if "secrets" in container:
        for secret in container["secrets"]:
            secret_name = secret["name"]
            secret_arn = secret["valueFrom"]
            secret_value = get_resolved_secret(secret_arn)
            print(f"{secret_name}={secret_value}")


def select_task_definition():
    # List all task definitions, sorted by descending order (newest first)
    paginator = ecs_client.get_paginator("list_task_definitions")
    task_definitions = []
    for page in paginator.paginate(sort="DESC"):
        task_definitions.extend(page["taskDefinitionArns"])

    # Filter to keep only the latest version of each task definition family
    unique_families = set()
    latest_task_defs = []
    for task_def in task_definitions:
        family = task_def.split("/")[-1].split(":")[0]  # Extract family name
        if family not in unique_families:
            unique_families.add(family)
            latest_task_defs.append(task_def)

    selected_task_def = pzp(candidates=latest_task_defs, lazy=True)

    if not selected_task_def:
        raise ValueError("No task definition selected. Exiting.")

    return selected_task_def


def select_container(task_definition):
    """Select a container from the task definition."""
    # Get the task definition details
    response = ecs_client.describe_task_definition(taskDefinition=task_definition)
    containers = response["taskDefinition"]["containerDefinitions"]

    container_names = [container["name"] for container in containers]

    selected_container = pzp(candidates=container_names, lazy=True)

    if not selected_container:
        raise ValueError("No container selected. Exiting.")

    # Find the selected container
    for container in containers:
        if container["name"] == selected_container:
            return container


def main():
    """Main function to interactively select a task definition and container."""
    try:
        task_definition = select_task_definition()

        container = select_container(task_definition)

        print_container_details(container)
    except Exception as e:
        print(f"Error: {e}", file=stderr)
        exit(1)


if __name__ == "__main__":
    main()