Quick and dirty script to clean up various types of resources in an AWS account

aws_cleanup.py

#!/usr/bin/env python3

import argparse
import boto3

# Cleanup misc AWS resources
# NOTE: this script does not currently handle pagination

def get_regions():
    ec2 = boto3.client("ec2")
    regions = ec2.describe_regions()
    return [r["RegionName"] for r in regions["Regions"]]

def cleanup_delivery_channels():
    print("Cleaning up delivery channels")
    for region in get_regions():
        config_client = boto3.client("config", region)
        delivery_channels = config_client.describe_delivery_channels()
        for channel in delivery_channels["DeliveryChannels"]:
            proceed = input("Delete {} (y/n)? ".format(channel["name"]))
            if proceed == "y":
                config_client.delete_delivery_channel(DeliveryChannelName=channel["name"])

def cleanup_sns():
    print("Cleaning up SNS topics")
    sns_client = boto3.client("sns")
    result = sns_client.list_topics()
    topic_arns = [topic["TopicArn"] for topic in result["Topics"]]
    while "NextToken" in result:
        result = sns_client.list_topics(NextToken=result["NextToken"])
        topic_arns.extend([topic["TopicArn"] for topic in result["Topics"]])

    for arn in topic_arns:
        sns_client.delete_topic(TopicArn=arn)


def cleanup_secrets_manager():
    print("Cleaning up SecretsManager secrets")
    for region in get_regions():
        sm_client = boto3.client("secretsmanager", region)
        result = sm_client.list_secrets()
        secrets = [secret["ARN"] for secret in result["SecretList"]]
        while "NextToken" in result:
            result = sm_client.list_secrets(NextToken=result["NextToken"])
            secrets.extend([secret["ARN"] for secret in result["SecretList"]])

        for arn in secrets:
            sm_client.delete_secret(SecretId=arn, ForceDeleteWithoutRecovery=True)


def cleanup_cloudtrail():
    print("Cleaning up CloudTrail Trails")
    cloudtrail_client = boto3.client("cloudtrail")
    trails = cloudtrail_client.list_trails()
    for trail in trails["Trails"]:
        trail_name = trail["Name"]
        trail_region = trail["HomeRegion"]
        proceed = input("Delete {} (y/n)? ".format(trail_name))
        if proceed == "y":
            regional_client = boto3.client("cloudtrail", trail_region)
            regional_client.delete_trail(Name=trail_name)


def cleanup_iam_role_policies(role_name):
    print("Cleaning up IAM role policies")
    iam = boto3.client("iam")
    role_policies = iam.list_role_policies(RoleName=role_name)
    for role_policy in role_policies["PolicyNames"]:
        iam.delete_role_policy(
                PolicyName=role_policy,
                RoleName=role_name
        )


def cleanup_attached_policies(role_name):
    print("Cleaning up attached IAM policies")
    iam = boto3.client("iam")
    attached_policies = iam.list_attached_role_policies(RoleName=role_name)
    for attached_policy in attached_policies["AttachedPolicies"]:
        iam.detach_role_policy(
                PolicyArn=attached_policy["PolicyArn"],
                RoleName=role_name
        )


def cleanup_config_recorders():
    print("Cleaning up config recorders")
    for region in get_regions():
        config_client = boto3.client("config", region)
        recorders = config_client.describe_configuration_recorders()
        for recorder in recorders["ConfigurationRecorders"]:
            proceed = input("Delete {} (y/n)? ".format(recorder["name"]))
            if proceed == "y":
                config_client.delete_configuration_recorder(ConfigurationRecorderName=recorder["name"])


def remove_role_from_instance_profile(role):
    print("Cleaning up IAM roles")
    iam = boto3.client("iam")
    response = iam.ListInstanceProfilesForRole(RoleName=role)
    for profile in response["InstanceProfiles"]:
        response = iam.remove_role_from_instance_profile(
            InstanceProfileName=profile["InstanceProfileName"],
            RoleName=role,
        )

def cleanup_iam_roles():
    print("Cleaning up IAM roles")
    iam = boto3.client("iam")
    roles = iam.list_roles()
    for role in roles["Roles"]:
        role_name = role["RoleName"]
        _input = input("Delete {} (y/n)? ".format(role_name))
        if _input != "y":
            continue
        cleanup_iam_role_policies(role_name)
        cleanup_attached_policies(role_name)
        remove_role_from_instance_profile(role_name)
        iam.delete_role(RoleName=role_name)

def cleanup_s3():
    print("Cleaning up S3 buckets")
    s3 = boto3.client("s3")
    response = s3.list_buckets()
    for bucket in response["Buckets"]:
        _input = input("Delete {} (y/n)? ".format(bucket["Name"]))
        if _input == "y":
            delete_bucket(bucket)

def delete_bucket(bucket):
    s3 = boto3.resource("s3")
    bucket_resource = s3.Bucket(bucket["Name"])
    bucket_resource.object_versions.delete()
    s3Client = boto3.client("s3")
    s3Client.delete_bucket(Bucket=bucket["Name"])

def cleanup_guardduty_detectors(non_interactive):
    print("Cleaning up GuardDuty detectors")
    for region in get_regions():
        gd = boto3.client("guardduty", region)
        detectors = gd.list_detectors()
        for detector in detectors["DetectorIds"]:
            if not non_interactive:
                proceed = input("Delete detector {} (y/n)? ".format(detector))
                if proceed != "y":
                    continue
            gd.delete_detector(DetectorId=detector)


parser = argparse.ArgumentParser()
parser.add_argument(
    "--force",
    action="store_true",
    help="If included, do not prompt for confirmation. WARNING: Automatically cleans up all given resources!"
)
parser.add_argument(
    "--s3",
    action="store_true",
    help="Whether to clean up S3 buckets by first emptying the bucket (delete all objects, all versions) and then deleting the bucket"
)
parser.add_argument(
    "--delivery-channels",
    action="store_true",
    help="Whether to clean up AWS Config Delivery Channels in all regions"
)
parser.add_argument(
    "--guardduty-detectors",
    action="store_true",
    help="Whether to clean up GuardDuty detectors in all regions"
)
parser.add_argument(
    "--iam-roles",
    action="store_true",
    help="Whether to clean up IAM roles by detaching all policies, deleting inline policies, and then deleting the role"
)
parser.add_argument(
    "--cloudtrail",
    action="store_true",
    help="Whether to clean up CloudTrail Trails in all regions"
)
parser.add_argument(
    "--config-recorders",
    action="store_true",
    help="Whether to clean up AWS Config Configuration Recorders in all regions"
)
parser.add_argument(
    "--sns",
    action="store_true",
    help="Whether to clean up AWS SNS topics in all regions"
)
parser.add_argument(
    "--secrets-manager",
    action="store_true",
    help="Whether to clean up AWS Secrets Manager Secrets in all regions"
)
parser.add_argument(
    "--all",
    action="store_true",
    help="Whether or not to clean up all supported resources"
)
parser.add_argument(
    "--non-interactive",
    action="store_true",
    help="If provided, will not prompt for confirmation"
)
args = parser.parse_args()

if __name__ == "__main__":
    if args.s3 or args.all:
        cleanup_s3()

    if args.config_recorders or args.all:
        cleanup_config_recorders()

    if args.delivery_channels or args.all:
        cleanup_delivery_channels()

    if args.iam_roles or args.all:
        cleanup_iam_roles()

    if args.guardduty_detectors or args.all:
        cleanup_guardduty_detectors(args.non_interactive)

    if args.cloudtrail or args.all:
        cleanup_cloudtrail()

    if args.sns or args.all:
        cleanup_sns()

    if args.secrets_manager or args.all:
        cleanup_secrets_manager()