del_vpc.py

#!/usr/bin/env python3

#### borrowed heavily from https://stackoverflow.com/questions/53519058/how-to-delete-vpc-with-all-its-dependencies-using-boto3
#### also found this online: https://github.com/jeffbrl/aws-vpc-destroy/blob/master/vpc_destroy.py

import boto3
import json
import re
import sys
from pprint import pprint
import argparse
import time

dry_run = True


def get_name_from_tags(tags):
    for t in tags:
        if t["Key"] == "Name":
            return t["Value"]
    return ""


def del_igws(vpc):
    igws = vpc.internet_gateways.all()
    if not igws:
        return
    for igw in igws:
        print("    - detaching igw {}".format(igw.id))
        if not dry_run:
            igw.detach_from_vpc(VpcId=vpc.id)
        print("    - deleting igw {}".format(igw.id))
        if not dry_run:
            igw.delete()


def del_vpc_endpoints(vpc):
    c = boto3.client('ec2')
    response = c.describe_vpc_endpoints(Filters=[{'Name': 'vpc-id', 'Values':[ vpc.id ]}])
    for ep in response['VpcEndpoints']:
        #print("endpoint")
        #pprint(services)
        print("    - deleting vpc endpoint {} ({})".format(ep["VpcEndpointId"], ep["VpcEndpointType"]))
        if not dry_run:
            response = c.delete_vpc_endpoints(VpcEndpointIds=[ep["VpcEndpointId"]])
            if len(response["Unsuccessful"]) > 0:
                raise Exception("Error deleting vpc endpoint {} ({}): Error code {}, {}", ep["VpcEndpointId"], ep["VpcEndpointType"], response["Unsuccessful"][0]["Error"]["Code"], response["Unsuccessful"][0]["Error"]["Message"]) 


def del_tgw_attachments(vpc):
    c = boto3.client('ec2')
    response = c.describe_transit_gateway_vpc_attachments(Filters=[{'Name': 'vpc-id', 'Values':[ vpc.id ]}])
    for tgwa in response['TransitGatewayVpcAttachments']:
        if tgwa["State"] == "deleted":
            print("    - TGW attachment {} is already marked as deleted".format(tgwa["TransitGatewayAttachmentId"]))
            continue

        print("    - deleting TGW attachment {}".format(tgwa["TransitGatewayAttachmentId"]))
        if not dry_run:
            response = c.delete_transit_gateway_vpc_attachment(TransitGatewayAttachmentId=tgwa['TransitGatewayAttachmentId'])
            #### HACK! wait for the ENIs to be cleaned up; we should really implement a smart check for them, 
            #### but I'm sick of writing this damn script tonight
            time.sleep(60)


def del_network_interfaces(vpc):
    network_interfaces = vpc.network_interfaces.all()
    if not network_interfaces:
        return
    for network_interface in network_interfaces: 
        print("    - detaching network interface {}".format(network_interface.id))
        if not dry_run:
            network_interface.detach()
        print("    - deleting network interface {}".format(network_interface.id))
        if not dry_run:
            network_interface.delete()


def is_main_route_table(route_table):
    for aa in route_table.associations_attribute:
        if aa.get('Main') == True:
            return True
    return False


def del_route_table_associations(route_table):
    route_table_associations = route_table.associations
    if not route_table_associations:
        return
    for rta in route_table_associations:
        print("      - deleting route table association {}".format(rta.id))
        if not dry_run:
            rta.delete()
        

def del_route_tables(vpc):
    route_tables = vpc.route_tables.all()
    if not route_tables:
        return
    for route_table in route_tables:
        name = get_name_from_tags(route_table.tags)
        if is_main_route_table(route_table):
            print("    - route table {} ({}) is the main route table; not deleting".format(route_table.id, name))
            continue

        print("    - removing route table associations for route table {} ({})".format(route_table.id, name))
        del_route_table_associations(route_table)
        print("    - removing route table {} ({})".format(route_table.id, name))
        if not dry_run:
            route_table.delete()


def del_subnets(vpc):
    subnets = vpc.subnets.all()
    if not subnets:
        return
    for subnet in subnets: 
        name = get_name_from_tags(subnet.tags)
        print("    - deleting subnet: {} ({})".format(subnet.id, name))
        if not dry_run:
            subnet.delete()


def del_nacls(vpc):
    nacls = vpc.network_acls.all()
    if not nacls:
        return
    for nacl in nacls: 
        name = get_name_from_tags(nacl.tags)
        if nacl.is_default:
            print("    - NACL {} ({}) is default; not deleting".format(nacl.id, name))
            continue

        print("    - removing NACL {} ({})".format(nacl.id, name))
        if not dry_run:
            nacl.delete()


def del_security_groups(vpc):
    security_groups = vpc.security_groups.all()
    if not security_groups:
        return
    for sg in security_groups: 
        name = get_name_from_tags(sg.tags)
        if sg.group_name == "default":
            print("    - security group {} ({}) is the default; not deleting".format(sg.id, name))
            continue

        print("    - removing security group {} ({})".format(sg.id, name))
        if not dry_run:
            sg.delete()


def del_vpc(id):
    try:
        ec2 = boto3.resource('ec2')
        vpc = ec2.Vpc(id)
        name = get_name_from_tags(vpc.tags)
        print("Removing prerequisites for vpc {} ({})".format(vpc.id, name))
        print("  - Detaching and removing internet gateways...")
        del_igws(vpc)
        print("  - Deleting route tables...")
        del_route_tables(vpc)
        print("  - Deleting VPC endpoints...")
        del_vpc_endpoints(vpc)
        print("  - Deleting transit gateway attachments...")
        del_tgw_attachments(vpc)
        print("  - Deleting network interfaces...")
        del_network_interfaces(vpc)
        print("  - Deleting subnets...")
        del_subnets(vpc)
        print("  - Deleting NACLs...")
        del_nacls(vpc)
        print("  - Deleting security groups...")
        del_security_groups(vpc)

        print("Removing vpc {} ({})".format(vpc.id, name))
        if not dry_run:
            vpc.delete()

    except boto3.exceptions.Boto3Error as e:
        print(e)
        print("Please remove dependencies and delete VPC manually.")


def str2bool(v):
    if isinstance(v, bool):
        return v
    if v.lower() in ('yes', 'true', 't', 'y', '1'):
        return True
    elif v.lower() in ('no', 'false', 'f', 'n', '0'):
        return False
    else:
        raise argparse.ArgumentTypeError('Boolean value expected.')

parser = argparse.ArgumentParser("del_vpc")
parser.add_argument("--vpc-id", help="The ID of the VPC to be deleted.", required=True)
parser.add_argument("--dry-run", help="whether to perform a dry run or actually delete the VPC (defaults to True)", default=True)
args = parser.parse_args()

dry_run = str2bool(args.dry_run)
vpc_id = args.vpc_id

if not dry_run:
    if input("This will delete the specified VPC and all of its constituent resources.  It cannot be undone.\nAre you sure you want to proceed? (y/n) ") != "y":
        print("Exiting.")
        exit()

del_vpc(vpc_id)