Amazon Virtual Private Cloud (Amazon VPC) is a core virtual networking service provided by Amazon Web Service to manage each account’s infrastructure within the Amazon Cloud. AWS VPC allows you to create a virtual private data center and private network within your AWS account. Each VPC is logically isolated from every other VPC within the AWS network. This Boto3 VPC Python tutorial covers managing VPCs, Subnets, Security Groups, ACLs, Internet, NAT gateways, and Route Tables using Python AWS SDK.

Table of contents

Prerequisites

To start managing Amazon VPC programmatically through the API, you must configure your Python environment.

In general, here’s what you need to have installed:

  • Python 3
  • Boto3
  • AWS CLI tools

Alternatively, you can use a Cloud9 IDE.

What is an Amazon VPC?

Amazon VPC is a virtual data center within the AWS infrastructure that gives you complete control over your virtual networking environment, including resource placement, connectivity, and security. At the same time, it removes the overhead and complexity of setting up a data center with cables, server racks, hardware, power supply, etc.

Each VPC you create is logically isolated from other virtual networks in the AWS cloud and fully customizable based on business and application requirements. You can choose the IP address range, create subnets, configure route tables, set up network gateways, define security rules using security groups, network access control lists, and many more.

Connect to Amazon VPC using Boto3

The Boto3 library provides you with two ways to access APIs for managing AWS services:

  • The client allows you to access the low-level API data. For example, you can get access to API response data in JSON format.
  • The resource allows you to use AWS services in a higher-level object-oriented way. For more information on the topic, take a look at AWS CLI vs botocore vs Boto3.

Here’s how you can instantiate the Boto3 EC2 client to start working with Amazon VPC APIs:

import boto3
AWS_REGION = "us-east-2"
vpc_client = boto3.client("ec2", region_name=AWS_REGION)

Similarly, you can instantiate the Boto3 EC2 resource to interact with Amazon VPC APIs:

import boto3
AWS_REGION = "us-east-2"
sqs_resource = boto3.resource("ec2", region_name=AWS_REGION)

How to manage VPCs using Boto3?

When creating a VPC, you set up a data center in the cloud, utilizing excellent AWS infrastructure. It’s secure, highly available, and scalable.

Boto3 library provides ec2 client and ec2 resource objects to interact with the Amazon VPC. Using these two objects, you can manage, create, update and delete the VPCs.

Create a VPC

Every AWS account comes with a default VPC that is pre-configured with all the required resources so you can start using the default VPC and deploy your resources in the VPC. Apart from the default VPC, you can also create custom VPCs. Custom VPCs give more control over the configurations.

Creating a Default VPC

To create default VPC, you need to use the create_default_vpc() method of the Boto3 EC2 client.

If you deleted your default VPC, you could re-create it later. As Amazon VPC is a regional service, you cannot have more than one default VPC per Region.

import logging
import boto3
from botocore.exceptions import ClientError
import json
AWS_REGION = 'us-east-2'
# logger config
logger = logging.getLogger()
logging.basicConfig(level=logging.INFO,
                    format='%(asctime)s: %(levelname)s: %(message)s')
vpc_client = boto3.client("ec2", region_name=AWS_REGION)

def create_default_vpc():
    """
    Creates a default VPC in the configured availability zone.
    """
    try:
        response = vpc_client.create_default_vpc()
    except ClientError:
        logger.exception('Could not create default vpc.')
        raise
    else:
        return response

if __name__ == '__main__':
    logger.info(f'Creating default VPC...')
    default_vpc = create_default_vpc()
    default_vpc_id = default_vpc['Vpc']['VpcId']
    logger.info(f'Default VPC is created with VPC ID: {default_vpc_id}')

The create_default_vpc() method does not require any argument. This method will only return successful output if no default VPC is present in a configured availability zone.

By default, the create_default_vpc() method will create a VPC with IP range CIDR 172.31.0.0/16 and a default subnet in each availability zone. We will look at subnets in the next section.

The create_default_vpc() method returns a dictionary object containing all the information about the newly created default VPC.

Here’s a code execution output.

Boto3 VPC - Creating a Default VPC
Creating a Default VPC

Creating a Custom VPC

When you need to take more control over the networking attributes of a VPC, you can create a custom VPC. Custom VPC provides more granular control over the configurations such as:

  • IP address range
  • Private and public subnets
  • Network security settings

To create a custom VPC, you can use the create_vpc() method from the Boto3 EC2 resource. In the example below, we are creating the custom VPC.

To read more about the Boto3 client and Boto3 resource, check out our article Introduction to the Boto3 library.

import logging
import boto3
from botocore.exceptions import ClientError
import json
AWS_REGION = 'us-east-2'
# logger config
logger = logging.getLogger()
logging.basicConfig(level=logging.INFO,
                    format='%(asctime)s: %(levelname)s: %(message)s')
vpc_resource = boto3.resource("ec2", region_name=AWS_REGION)

def create_custom_vpc(ip_cidr):
    """
    Creates a custom VPC with the specified configuration.
    """
    try:
        response = vpc_resource.create_vpc(CidrBlock=ip_cidr,
                                           InstanceTenancy='default',
                                           TagSpecifications=[{
                                               'ResourceType':
                                               'vpc',
                                               'Tags': [{
                                                   'Key':
                                                   'Name',
                                                   'Value':
                                                   'hands-on-cloud-custom-vpc'
                                               }]
                                           }])
    except ClientError:
        logger.exception('Could not create a custom vpc.')
        raise
    else:
        return response

if __name__ == '__main__':
    # Constants
    IP_CIDR = '192.168.0.0/16'
    logger.info(f'Creating a custom VPC...')
    custom_vpc = create_custom_vpc(IP_CIDR)
    logger.info(f'Custom VPC is created with VPC ID: {custom_vpc.id}')

The required argument is:

  • CidrBlock: The IPv4 networking range for the VPC specified in CIDR notation. For example, 192.168.0.0/16.

Optional arguments used in the above example:

  • InstanceTenancy: The instance tenancy options for EC2 instances deployed into the VPC. For default, instances are launched with shared tenancy by default. Other options include dedicated and host tenancy.
  • TagSpecifications: The resource tags to assign to the VPC.

The create_vpc() method returns a Vpc resource object containing the VPC Id.

Here’s the code execution output:

Creating a Custom VPC
Creating a Custom VPC

Describe VPCs

To get detailed information about the existing VPCs in your AWS account, you need to use the describe_vpcs() method from the Boto3 library. This method allows you to describe existing VPCs by using either VPC Ids or Filters. We will use the Filters argument in the example below to find the required VPC by the Tag.

We will use the Boto3 library paginator object to get the complete output from the describe_vpcs() method.

Some AWS requests return incomplete output, therefore, require subsequent requests to get the complete result. The process of sending subsequent requests to continue where a previous request left off is called pagination.

import logging
import boto3
from botocore.exceptions import ClientError
import json
AWS_REGION = 'us-east-2'
# logger config
logger = logging.getLogger()
logging.basicConfig(level=logging.INFO,
                    format='%(asctime)s: %(levelname)s: %(message)s')
vpc_client = boto3.client("ec2", region_name=AWS_REGION)

def describe_vpcs(tag, tag_values, max_items):
    """
    Describes one or more of your VPCs.
    """
    try:
        # creating paginator object for describe_vpcs() method
        paginator = vpc_client.get_paginator('describe_vpcs')
        # creating a PageIterator from the paginator
        response_iterator = paginator.paginate(
            Filters=[{
                'Name': f'tag:{tag}',
                'Values': tag_values
            }],
            PaginationConfig={'MaxItems': max_items})
        full_result = response_iterator.build_full_result()
        vpc_list = []
        for page in full_result['Vpcs']:
            vpc_list.append(page)
    except ClientError:
        logger.exception('Could not describe VPCs.')
        raise
    else:
        return vpc_list

if __name__ == '__main__':
    # Constants
    TAG = 'Name'
    TAG_VALUES = ['hands-on-cloud-custom-vpc']
    MAX_ITEMS = 10
    vpcs = describe_vpcs(TAG, TAG_VALUES, MAX_ITEMS)
    logger.info('VPC Details: ')
    for vpc in vpcs:
        logger.info(json.dumps(vpc, indent=4) + '\n')

The arguments used in the above example:

  • Filters: Specifies the different available filters for narrowing the VPCs search criteria. For all available filters, refer here.
  • PaginationConfig: Specifies the configuration for the Paginator object. We have used the MaxItems attribute, which specifies the maximum number of VPCs to return in one request.

The describe_vpcs() method returns a python dictionary object as an output.

Here’s a code execution output:

3. Describe VPCs
Describe VPCs

Describe VPC Attributes

To get the specific VPC attribute information, you can use the describe_vpc_attribute() method from the Boto3 library.

The describe_vpc_attribute() method supports two attributes enableDnsSupport & enableDnsHostnames.

import logging
import boto3
from botocore.exceptions import ClientError
import json
AWS_REGION = 'us-east-2'
# logger config
logger = logging.getLogger()
logging.basicConfig(level=logging.INFO,
                    format='%(asctime)s: %(levelname)s: %(message)s')
vpc_client = boto3.client("ec2", region_name=AWS_REGION)

def describe_vpc_attribute(vpc_id, attribute):
    """
    Describes the specified attribute of the specified VPC.
    """
    try:
        response = vpc_client.describe_vpc_attribute(Attribute=attribute,
                                                     VpcId=vpc_id)
    except ClientError:
        logger.exception('Could not describe a vpc attribute.')
        raise
    else:
        return response

if __name__ == '__main__':
    # Constants
    VPC_ID = 'vpc-0c588f5fd7bfb4534'
    ATTRIBUTE = 'enableDnsSupport'
    custom_vpc_attribute = describe_vpc_attribute(VPC_ID, ATTRIBUTE)
    logger.info(
        f'VPC attribute details: \n{json.dumps(custom_vpc_attribute, indent=4)}'
    )

Required attributes are:

  • Attribute: The VPC attribute. The describe_vpc_attribute() method supports two attributes enableDnsSupport and enableDnsHostnames.
  • VpcId: The ID of the VPC.

The describe_vpc_attribute() method returns a Python dictionary object as an output.

Here’s the code execution output:

Describe VPC Attributes
Describe VPC Attributes

Modify VPC Attributes

To modify the existing VPC attribute’s value, you can use the modify_vpc_attribute() method from the Boto3 library.

import logging
import boto3
from botocore.exceptions import ClientError
import json
AWS_REGION = 'us-east-2'
# logger config
logger = logging.getLogger()
logging.basicConfig(level=logging.INFO,
                    format='%(asctime)s: %(levelname)s: %(message)s')
vpc_client = boto3.client("ec2", region_name=AWS_REGION)

def modify_vpc_attribute(vpc_id, dns_support):
    """
    Modifies the specified attribute of the specified VPC.
    """
    try:
        response = vpc_client.modify_vpc_attribute(
            EnableDnsSupport={'Value': dns_support}, VpcId=vpc_id)
    except ClientError:
        logger.exception('Could not modify the vpc attribute.')
        raise
    else:
        return response

if __name__ == '__main__':
    # Constants
    VPC_ID = 'vpc-0bd00c7e9d953cb23'
    enableDnsSupport = True
    vpc_attribute = modify_vpc_attribute(VPC_ID, enableDnsSupport)
    logger.info(
        f'VPC attribute enableDnsSupport modified. New value: {enableDnsSupport}'
    )

The required attribute is:

  • VpcId: The ID of the VPC.

An optional argument used in the above example is:

  • EnableDnsSupport: Specifies whether the DNS resolution is supported for the VPC.

The modify_vpc_attribute() method supports two arguments, enableDnsSupport & enableDnsHostnames. However, you can modify one argument at a time only.

The modify_vpc_attribute() does not return any output on a successful request.

Here’s a code execution output:

Modify VPC Attributes
Modify VPC Attributes

Delete a VPC

To delete the existing VPC, you can use the delete_vpc() method from the Boto3 library.

All the resources attached with the VPC must be detached OR deleted before deleting the VPC.

import logging
import boto3
from botocore.exceptions import ClientError
import json
AWS_REGION = 'us-east-2'
# logger config
logger = logging.getLogger()
logging.basicConfig(level=logging.INFO,
                    format='%(asctime)s: %(levelname)s: %(message)s')
vpc_client = boto3.client("ec2", region_name=AWS_REGION)

def delete_vpc(vpc_id):
    """
    Deletes the specified VPC.
    """
    try:
        response = vpc_client.delete_vpc(VpcId=vpc_id)
    except ClientError:
        logger.exception('Could not delete the vpc.')
        raise
    else:
        return response

if __name__ == '__main__':
    # Constants
    VPC_ID = 'vpc-0bd00c7e9d953cb23'
    vpc = delete_vpc(VPC_ID)
    logger.info(f'VPC {VPC_ID} is deleted successfully.')

The required attribute is:

  • VpcId: The ID of the VPC.

The delete_vpc() method does not return any output on a successful request.

Here’s the code execution output:

Delete a VPC
Delete a VPC

How to manage Subnets using Boto3?

Every VPC has to have at least one subnet before you can start using it.

There are two types of subnets:

  • Private subnet is only accessible within the VPC resources and can not be accessed from the Internet.
  • Public subnet contains a route to 0.0.0.0/0 network through the Internet Gateway; resources in the Public subnet can be exposed to the Internet.

Create a Subnet

Similar to the VPCs, Subnets can be one of two types, default subnet or custom subnet.

The default subnet can be created in the default VPC only.

Custom subnets give more control over the networking configuration.

Creating a Default Subnet

To create the default subnet in the default VPC, you can use the create_default_subnet() method from the Boto3 library.

import logging
import boto3
from botocore.exceptions import ClientError
import json
AWS_REGION = 'us-east-2'
# logger config
logger = logging.getLogger()
logging.basicConfig(level=logging.INFO,
                    format='%(asctime)s: %(levelname)s: %(message)s')
vpc_client = boto3.client("ec2", region_name=AWS_REGION)

def create_default_subnet(az):
    """
    Creates a default subnet in the configured availability zone.
    """
    try:
        response = vpc_client.create_default_subnet(AvailabilityZone=az)
    except ClientError:
        logger.exception('Could not create default subnet.')
        raise
    else:
        return response

if __name__ == '__main__':
    # Constants
    AvailabilityZone = 'us-east-2b'
    logger.info(f'Creating default subnet...')
    default_subnet = create_default_subnet(AvailabilityZone)
    logger.info(
        f'Default Subnet is created with ID: \n{json.dumps(default_subnet, indent=4)}'
    )

The required attribute is:

  • AvailabilityZone: The Availability Zone in which to create the default subnet.

The create_default_subnet() method returns the Python dictionary object containing the created subnet attributes.

Here’s the code execution output:

Creating a Default Subnet
Creating a Default Subnet

Creating a Custom Subnet

To create a custom subnet, you need to use the create_subnet() method from the Boto3 library.

In the below example, we are using the Boto3 EC2 resource to create the custom subnet.

import logging
import boto3
from botocore.exceptions import ClientError
import json
AWS_REGION = 'us-east-2'
# logger config
logger = logging.getLogger()
logging.basicConfig(level=logging.INFO,
                    format='%(asctime)s: %(levelname)s: %(message)s')
vpc_resource = boto3.resource("ec2", region_name=AWS_REGION)

def create_custom_subnet(az, vpc_id, cidr_block):
    """
    Creates a custom subnet with the specified configuration.
    """
    try:
        response = vpc_resource.create_subnet(TagSpecifications=[
            {
                'ResourceType': 'subnet',
                'Tags': [{
                    'Key': 'Name',
                    'Value': 'hands-on-cloud-custom-subnet'
                }]
            },
        ],
                                              AvailabilityZone=az,
                                              VpcId=vpc_id,
                                              CidrBlock=cidr_block)
    except ClientError:
        logger.exception(f'Could not create a custom subnet.')
        raise
    else:
        return response

if __name__ == '__main__':
    # Constants
    CIDR_BLOCK = '192.168.1.0/20'
    VPC_ID = 'vpc-048604f523ad01d74'
    AZ = 'us-east-2a'
    logger.info(f'Creating a custom Subnet...')
    custom_subnet = create_custom_subnet(AZ, VPC_ID, CIDR_BLOCK)
    logger.info(f'Custom Subnet is created with Subnet ID: {custom_subnet.id}')

The required attributes are:

  • VpcId: The ID of the VPC.
  • CidrBlock: The IPv4 networking range for the Subnet specified in CIDR notation. For example, 192.168.1.0/16.

Optional arguments used in the above example:

  • TagSpecifications: Resource Tags for the Subnet.
  • AvailabilityZone: The Availability Zone or Local Zone for the Subnet.

The create_subnet() method returns the Subnet resource object after a successful request.

Here’s a code execution output:

Creating a Custom Subnet
Creating a Custom Subnet

Describe Subnets

To get the attributes of the existing Subnets in your AWS account, you need to use the describe_subnets() method from the Boto3 library. This method allows you to describe existing Subnets by using Subnet Ids and Filters. We will use Filters in the example below.

We will use the Boto3 library paginator object to get the complete output from the describe_subnets() method.

import logging
import boto3
from botocore.exceptions import ClientError
import json
AWS_REGION = 'us-east-2'
# logger config
logger = logging.getLogger()
logging.basicConfig(level=logging.INFO,
                    format='%(asctime)s: %(levelname)s: %(message)s')
vpc_client = boto3.client("ec2", region_name=AWS_REGION)

def describe_subnets(tag, tag_values, max_items):
    """
    Describes one or more of your Subnets.
    """
    try:
        # creating paginator object for describe_subnets() method
        paginator = vpc_client.get_paginator('describe_subnets')
        # creating a PageIterator from the paginator
        response_iterator = paginator.paginate(
            Filters=[{
                'Name': f'tag:{tag}',
                'Values': tag_values
            }],
            PaginationConfig={'MaxItems': max_items})
        full_result = response_iterator.build_full_result()
        subnet_list = []
        for page in full_result['Subnets']:
            subnet_list.append(page)
    except ClientError:
        logger.exception('Could not describe subnets.')
        raise
    else:
        return subnet_list

if __name__ == '__main__':
    # Constants
    TAG = 'Name'
    TAG_VALUES = ['hands-on-cloud-custom-subnet']
    MAX_ITEMS = 10
    subnets = describe_subnets(TAG, TAG_VALUES, MAX_ITEMS)
    logger.info('Subnet Details: ')
    for subnet in subnets:
        logger.info(json.dumps(subnet, indent=4) + '\n')

Optional arguments used in the above example:

  • Filters: Specifies the different available filters for narrowing the Subnets search criteria. For all available filters, refer here.
  • PaginationConfig: Specifies the configuration for the Paginator object. We have used the MaxItems attribute, which specifies the maximum number of Subnets to return in one request.

The describe_subnets() method returns a python dictionary object as an output.

Here’s a code execution output:

Describe Subnets
Describe Subnets

Modify Subnet Attributes

To modify the existing subnet attribute’s value, you need to use the modify_subnet_attribute() method from the Boto3 library.

import logging
import boto3
from botocore.exceptions import ClientError
import json
AWS_REGION = 'us-east-2'
# logger config
logger = logging.getLogger()
logging.basicConfig(level=logging.INFO,
                    format='%(asctime)s: %(levelname)s: %(message)s')
vpc_client = boto3.client("ec2", region_name=AWS_REGION)

def modify_subnet_attribute(subnet_id, map_public_ip_on_launch):
    """
    Modifies the specified attribute of the specified subnet.
    """
    try:
        response = vpc_client.modify_subnet_attribute(
            MapPublicIpOnLaunch={'Value': map_public_ip_on_launch},
            SubnetId=subnet_id)
    except ClientError:
        logger.exception('Could not modify the Subnet attribute.')
        raise
    else:
        return response

if __name__ == '__main__':
    # Constants
    SUBNET_ID = 'subnet-071923fde0da8166e'
    MAP_PUBLIC_IP_ON_LAUNCH = True
    subnet_attribute = modify_subnet_attribute(SUBNET_ID,
                                               MAP_PUBLIC_IP_ON_LAUNCH)
    logger.info(
        f'Subnet attribute MapPublicIpOnLaunch modified. New value: {MAP_PUBLIC_IP_ON_LAUNCH}'
    )

The required attribute is:

  • SubnetId: The ID of the Subnet.

An optional argument used in the above example:

  • MapPublicIpOnLaunch: Specifies if the specified Subnet should be assigned a public IPv4 address.

The modify_subnet_attribute() method supports few arguments,

  • AssignIpv6AddressOnCreation
  • MapPublicIpOnLaunch
  • MapCustomerOwnedIpOnLaunch
  • CustomerOwnedIpv4Pool

However, you can modify one argument at a time only.

The modify_subnet_attribute() does not return any output on a successful request.

Here’s an execution output:

Modify Subnet Attributes
Modify Subnet Attributes

Delete a Subnet

To delete the existing subnet, you need to use the delete_subnet() method from the Boto3 library.

All the resources running in the Subnet must be detached OR deleted before deleting the Subnet successfully.

import logging
import boto3
from botocore.exceptions import ClientError
import json
AWS_REGION = 'us-east-2'
# logger config
logger = logging.getLogger()
logging.basicConfig(level=logging.INFO,
                    format='%(asctime)s: %(levelname)s: %(message)s')
vpc_client = boto3.client("ec2", region_name=AWS_REGION)

def delete_subnet(subnet_id):
    """
    Deletes the specified Subnet.
    """
    try:
        response = vpc_client.delete_subnet(SubnetId=subnet_id)
    except ClientError:
        logger.exception('Could not delete the subnet.')
        raise
    else:
        return response

if __name__ == '__main__':
    # Constants
    SUBNET_ID = 'subnet-071923fde0da8166e'
    subnet = delete_subnet(SUBNET_ID)
    logger.info(f'Subnet {SUBNET_ID} is deleted successfully.')

The required attribute is:

  • SubnetId: The ID of the Subnet.

The delete_subnet() method does not return any output on a successful request.

Here’s the code execution output:

Delete a Subnet
Delete a Subnet

How to manage Security Groups using Boto3?

An AWS Security Group works as a virtual firewall for your EC2 instances and contains a set of rules that filter incoming and outgoing traffic. Both inbound and outbound rules control traffic flow to and from your instance for specified ports and protocols.

Create a Security Group

To create the security group, you need to use the create_security_group() method from the Boto3 library. In the below example, we are using the Boto3 EC2 resource to create the Security Group.

import logging
import boto3
from botocore.exceptions import ClientError
import json
AWS_REGION = 'us-east-2'
# logger config
logger = logging.getLogger()
logging.basicConfig(level=logging.INFO,
                    format='%(asctime)s: %(levelname)s: %(message)s')
vpc_resource = boto3.resource("ec2", region_name=AWS_REGION)

def create_security_group(description, groupname, vpc_id):
    """
    Creates a security group with the specified configuration.
    """
    try:
        response = vpc_resource.create_security_group(Description=description,
                                                      GroupName=groupname,
                                                      VpcId=vpc_id,
                                                      TagSpecifications=[{
                                                          'ResourceType':
                                                          'security-group',
                                                          'Tags': [{
                                                              'Key':
                                                              'Name',
                                                              'Value':
                                                              groupname
                                                          }]
                                                      }])
    except ClientError:
        logger.exception('Could not create a security group.')
        raise
    else:
        return response

if __name__ == '__main__':
    # Constants
    DESCRIPTION = 'Security group created for hands-on-cloud blog'
    GROUPNAME = 'hands-on-cloud-security-group'
    VPC_ID = 'vpc-048604f523ad01d74'
    logger.info(f'Creating a security group...')
    security_group = create_security_group(DESCRIPTION, GROUPNAME, VPC_ID)
    logger.info(f'Security group created with ID: {security_group.id}')

The required attributes are:

  • GroupName: The name of the security group.
  • Description: A description for the security group.

Optional arguments used in the above example:

  • VpcId: The VPC Id in which the security group is to be created.
  • TagSpecifications: AWS Resource tags for the security group.

The create_security_group() method returns a SecurityGroup resource object containing the Security Group Id.

Here’s a code execution output:

Create a Security Group
Create a Security Group

Manage Security Group Rules

Security Group rules define the network traffic parameters to control the traffic on the ports and protocol level. Every security group can have up to 50 rules. Security Group rules can also specify source IP addresses or an IP address range.

Create Security Group Ingress Rule

Ingress Rule defines the inbound traffic rules for a Security Group. To add an Ingress rule, you need to use the authorize_security_group_ingress() method from the Boto3 library.

import logging
import boto3
from botocore.exceptions import ClientError
import json
AWS_REGION = 'us-east-2'
# logger config
logger = logging.getLogger()
logging.basicConfig(level=logging.INFO,
                    format='%(asctime)s: %(levelname)s: %(message)s')
vpc_client = boto3.client("ec2", region_name=AWS_REGION)

def create_ingress_rule(security_group_id):
    """
    Creates a security group ingress rule with the specified configuration.
    """
    try:
        response = vpc_client.authorize_security_group_ingress(
            GroupId=security_group_id,
            IpPermissions=[{
                'IpProtocol': 'tcp',
                'FromPort': 80,
                'ToPort': 80,
                'IpRanges': [{
                    'CidrIp': '0.0.0.0/0'
                }]
            }, {
                'IpProtocol': 'tcp',
                'FromPort': 22,
                'ToPort': 22,
                'IpRanges': [{
                    'CidrIp': '0.0.0.0/0'
                }]
            }])
    except ClientError:
        logger.exception('Could not create ingress security group rule.')
        raise
    else:
        return response

if __name__ == '__main__':
    # Constants
    SECURITY_GROUP_ID = 'sg-08e4ffb1b4087e728'
    logger.info(f'Creating a security group ingress rule...')
    rule = create_ingress_rule(SECURITY_GROUP_ID)
    logger.info(
        f'Security group ingress rule created: \n{json.dumps(rule, indent=4)}')

The required attribute is:

  • GroupId:  Specifies the ID of the security group. You must specify either the security group ID or the security group name in the request.

An optional argument used in the above example:

  • IpPermissions: The sets of rules containing required IpProtocol, FromPort, ToPort, and IpRanges.

The authorize_security_group_ingress() method returns a Python dictionary object as an output.

Here’s the code execution output:

Create Security Group Ingress Rule
Create Security Group Ingress Rule

Create Security Group Egress Rule

Egress Rule defines the outbound traffic rules for a security group. To add an Egress rule, you need to use the authorize_security_group_egress() method from the Boto3 library.

import logging
import boto3
from botocore.exceptions import ClientError
import json
AWS_REGION = 'us-east-2'
# logger config
logger = logging.getLogger()
logging.basicConfig(level=logging.INFO,
                    format='%(asctime)s: %(levelname)s: %(message)s')
vpc_client = boto3.client("ec2", region_name=AWS_REGION)

def create_egress_rule(security_group_id):
    """
    Creates a security group egress rule with the specified configuration.
    """
    try:
        response = vpc_client.authorize_security_group_egress(
            GroupId=security_group_id,
            IpPermissions=[{
                'IpProtocol': 'tcp',
                'FromPort': 80,
                'ToPort': 80,
                'IpRanges': [{
                    'CidrIp': '0.0.0.0/0'
                }]
            }, {
                'IpProtocol': 'tcp',
                'FromPort': 22,
                'ToPort': 22,
                'IpRanges': [{
                    'CidrIp': '0.0.0.0/0'
                }]
            }])
    except ClientError:
        logger.exception('Could not create egress security group rule.')
        raise
    else:
        return response

if __name__ == '__main__':
    # Constants
    SECURITY_GROUP_ID = 'sg-08e4ffb1b4087e728'
    logger.info(f'Creating a security group egress rule...')
    rule = create_egress_rule(SECURITY_GROUP_ID)
    logger.info(
        f'Security group egress rule created: \n{json.dumps(rule, indent=4)}')

The required attribute is:

  • GroupId:  Specifies the ID of the security group. You must specify either the security group ID or the security group name in the request.

An optional argument used in the above example:

  • IpPermissions: The sets of rules containing required IpProtocol, FromPort, ToPort, and IpRanges.

The authorize_security_group_egress() method returns a Python dictionary object as an output.

Here’s a code execution output:

Create Security Group Egress Rule
Create Security Group Egress Rule

Modify Security Groups Rule

To modify the existing rule, you need to use the modify_security_group_rules() method from the Boto3 library.

import logging
import boto3
from botocore.exceptions import ClientError
import json
AWS_REGION = 'us-east-2'
# logger config
logger = logging.getLogger()
logging.basicConfig(level=logging.INFO,
                    format='%(asctime)s: %(levelname)s: %(message)s')
vpc_client = boto3.client("ec2", region_name=AWS_REGION)

def modify_rule(security_group_id, security_group_rule_id):
    """
    Modify the existing security group rules with the specified configuration.
    """
    try:
        response = vpc_client.modify_security_group_rules(
            GroupId=security_group_id,
            SecurityGroupRules=[
                {
                    'SecurityGroupRuleId': security_group_rule_id,
                    'SecurityGroupRule': {
                        'IpProtocol': 'tcp',
                        'FromPort': 8080,
                        'ToPort': 8080,
                        'CidrIpv4': '0.0.0.0/0',
                        'Description': 'hands-on-cloud-security-group-rules'
                    }
                },
            ])
    except ClientError:
        logger.exception('Could not modify security group rule.')
        raise
    else:
        return response

if __name__ == '__main__':
    # Constants
    SECURITY_GROUP_ID = 'sg-08e4ffb1b4087e728'
    SECURITY_GROUP_RULE_ID = 'sgr-0587b4899af1c7a2f'
    logger.info(f'Modifing a security group rule...')
    rule = modify_rule(SECURITY_GROUP_ID, SECURITY_GROUP_RULE_ID)
    logger.info(
        f'Security group rule modified: \n{json.dumps(rule, indent=4)}')

The required attributes are:

  • GroupId: Specifies the ID of the security group.
  • SecurityGroupRules: Security group properties to update. To view all the properties that you can modify, refer here.

In the above example, we are modifying the ingress rule created earlier with a different port number.

The modify_security_group_rules() method returns a Python dictionary object as an output.

Here’s the code execution output:

15. Modify Security Groups Rule
Modify Security Groups Rule

Delete Security Group Ingress Rule

To remove the existing Ingress rule from a Security Group, you need to use the revoke_security_group_ingress() method from the Boto3 library.

import logging
import boto3
from botocore.exceptions import ClientError
import json
AWS_REGION = 'us-east-2'
# logger config
logger = logging.getLogger()
logging.basicConfig(level=logging.INFO,
                    format='%(asctime)s: %(levelname)s: %(message)s')
vpc_client = boto3.client("ec2", region_name=AWS_REGION)

def delete_ingress_rule(security_group_id, security_group_rule_ids):
    """
    Deletes a security group ingress rule.
    """
    try:
        response = vpc_client.revoke_security_group_ingress(
            GroupId=security_group_id,
            SecurityGroupRuleIds=security_group_rule_ids)
    except ClientError:
        logger.exception('Could not delete ingress security group rule.')
        raise
    else:
        return response

if __name__ == '__main__':
    # Constants
    SECURITY_GROUP_ID = 'sg-08e4ffb1b4087e728'
    SECURITY_GROUP_RULE_ID = ['sgr-0587b4899af1c7a2f']
    logger.info(f'Removing a security group ingress rule(s)...')
    rule = delete_ingress_rule(SECURITY_GROUP_ID, SECURITY_GROUP_RULE_ID)
    logger.info(
        f'Security group ingress rule(s) deleted: \n{json.dumps(rule, indent=4)}'
    )

The required attributes are:

  • GroupId: Specifies the ID of the security group.
  • SecurityGroupRuleIds: Specifies the list of the Security Group Rule Ids that are to be deleted.

In the above example, we delete the ingress rule created earlier with the HTTP 8080 port number.

The revoke_security_group_ingress() method returns a python dictionary object as an output.

Here’s the code execution output:

Delete Security Group Ingress Rule
Delete Security Group Ingress Rule

Delete Security Group Egress Rule

To remove the existing Egress rule from a Security Group, you need to use the revoke_security_group_egress() method from the Boto3 library.

import logging
import boto3
from botocore.exceptions import ClientError
import json
AWS_REGION = 'us-east-2'
# logger config
logger = logging.getLogger()
logging.basicConfig(level=logging.INFO,
                    format='%(asctime)s: %(levelname)s: %(message)s')
vpc_client = boto3.client("ec2", region_name=AWS_REGION)

def delete_egress_rule(security_group_id, security_group_rule_ids):
    """
    Deletes a security group egress rule.
    """
    try:
        response = vpc_client.revoke_security_group_egress(
            GroupId=security_group_id,
            SecurityGroupRuleIds=security_group_rule_ids)
    except ClientError:
        logger.exception('Could not delete egress security group rule.')
        raise
    else:
        return response

if __name__ == '__main__':
    # Constants
    SECURITY_GROUP_ID = 'sg-08e4ffb1b4087e728'
    SECURITY_GROUP_RULE_ID = ['sgr-0c1b8c264ea2d0e6f']
    logger.info(f'Removing a security group egress rule(s)...')
    rule = delete_egress_rule(SECURITY_GROUP_ID, SECURITY_GROUP_RULE_ID)
    logger.info(
        f'Security group egress rule(s) deleted: \n{json.dumps(rule, indent=4)}'
    )

The required attributes are:

  • GroupId: Specifies the ID of the security group.
  • SecurityGroupRuleIds: Specifies the list of the Security Group Rule Ids that are to be deleted.

In the above example, we delete the egress rule created earlier with the SSH 22 port number.

The revoke_security_group_egress() method returns a Python dictionary object as an output.

Here’s the code execution output:

Delete Security Group Egress Rule
Delete Security Group Egress Rule

Describe Security Groups

To get the existing Security Groups details in your AWS account, you need to use the describe_security_groups() method from the Boto3 library. 

This method allows you to describe existing Security Groups using few methods, Security Group Ids, Security Group Names, and Filters. We will use Filters in the example below.

We will use the Boto3 library paginator object to get the complete output from the describe_security_groups() method.

import logging
import boto3
from botocore.exceptions import ClientError
import json
AWS_REGION = 'us-east-2'
# logger config
logger = logging.getLogger()
logging.basicConfig(level=logging.INFO,
                    format='%(asctime)s: %(levelname)s: %(message)s')
vpc_client = boto3.client("ec2", region_name=AWS_REGION)

def describe_security_groups(tag, tag_values, max_items):
    """
    Describes one or more of your security groups.
    """
    try:
        # creating paginator object for describe_subnets() method
        paginator = vpc_client.get_paginator('describe_security_groups')
        # creating a PageIterator from the paginator
        response_iterator = paginator.paginate(
            Filters=[{
                'Name': f'tag:{tag}',
                'Values': tag_values
            }],
            PaginationConfig={'MaxItems': max_items})
        full_result = response_iterator.build_full_result()
        security_groups_list = []
        for page in full_result['SecurityGroups']:
            security_groups_list.append(page)
    except ClientError:
        logger.exception('Could not describe Security Groups.')
        raise
    else:
        return security_groups_list

if __name__ == '__main__':
    # Constants
    TAG = 'Name'
    TAG_VALUES = ['hands-on-cloud-security-group']
    MAX_ITEMS = 10
    security_groups = describe_security_groups(TAG, TAG_VALUES, MAX_ITEMS)
    logger.info('Security Groups details: ')
    for security_group in security_groups:
        logger.info(json.dumps(security_groups, indent=4) + '\n')

Optional arguments used in the above example:

  • Filters: Specifies the different available filters for narrowing the Security Groups search criteria. For all available filters, refer here.
  • PaginationConfig: Specifies the configuration for the Paginator object. We have used the MaxItems attribute, which specifies the maximum number of Security Groups to return in one request.

Here’s a code execution output:

Describe Security Groups
Describe Security Groups

Describe Security Group Rules

To get the current Security Group Rules information in your AWS account, you need to use the describe_security_group_rules() method from the Boto3 library. 

This method allows you to describe existing Security Group Rules using two methods, Security Group Rule Ids and Filters. We will use Filters in the below example.

We will use the Boto3 library paginator object to get the complete output from the describe_security_group_rules() method.

import logging
import boto3
from botocore.exceptions import ClientError
import json
AWS_REGION = 'us-east-2'
# logger config
logger = logging.getLogger()
logging.basicConfig(level=logging.INFO,
                    format='%(asctime)s: %(levelname)s: %(message)s')
vpc_client = boto3.client("ec2", region_name=AWS_REGION)

def describe_security_groups_rules(security_group_ids, max_items):
    """
    Describes one or more of your security groups rules.
    """
    try:
        # creating paginator object for describe_subnets() method
        paginator = vpc_client.get_paginator('describe_security_group_rules')
        # creating a PageIterator from the paginator
        response_iterator = paginator.paginate(
            Filters=[{
                'Name': 'group-id',
                'Values': security_group_ids
            }],
            PaginationConfig={'MaxItems': max_items})
        full_result = response_iterator.build_full_result()
        security_groups_rules = []
        for page in full_result['SecurityGroupRules']:
            security_groups_rules.append(page)
    except ClientError:
        logger.exception('Could not describe Security Groups Rules.')
        raise
    else:
        return security_groups_rules

if __name__ == '__main__':
    # Constants
    SECURITY_GROUP_IDS = ['sg-08e4ffb1b4087e728']
    MAX_ITEMS = 10
    rules = describe_security_groups_rules(SECURITY_GROUP_IDS, MAX_ITEMS)
    logger.info('Security groups rules: ')
    for rule in rules:
        logger.info(json.dumps(rule, indent=4) + '\n')

Optional arguments used in the above example:

  • Filters: Specifies the different available filters for narrowing the Security Groups Rules search criteria. For all available filters, refer here.
  • PaginationConfig: Specifies the configuration for the Paginator object. We have used the MaxItems attribute, which specifies the maximum number of Security Groups Rules to return in one request.

In the above example, we have used group-id as a filter. Group-id specifies the list of security groups to get the ingress and egress rules.

Here’s the code execution output:

Describe Security Group Rules
Describe Security Group Rules

Delete Security Group

To delete the specific Security Group, you need to use the delete_security_group() method from the Boto3 library.

import logging
import boto3
from botocore.exceptions import ClientError
import json
AWS_REGION = 'us-east-2'
# logger config
logger = logging.getLogger()
logging.basicConfig(level=logging.INFO,
                    format='%(asctime)s: %(levelname)s: %(message)s')
vpc_client = boto3.client("ec2", region_name=AWS_REGION)

def delete_security_group(security_group_id):
    """
    Deletes the specified security group.
    """
    try:
        response = vpc_client.delete_security_group(GroupId=security_group_id)
    except ClientError:
        logger.exception('Could not delete the security group.')
        raise
    else:
        return response

if __name__ == '__main__':
    # Constants
    SECURITY_GROUP_ID = 'sg-08e4ffb1b4087e728'
    security_group = delete_security_group(SECURITY_GROUP_ID)
    logger.info(f'Security group {SECURITY_GROUP_ID} is deleted successfully.')

The required attribute is:

  • GroupId: Specifies the ID of the security group to be deleted.

The delete_security_group() method does not return any object on a successful request.

Here’s a code execution output:

Delete Security Group
Delete Security Group

How to manage Network Access Control Lists using Boto3?

Network Access Control Lists (or NACLs) provide an additional layer of network security to the AWS resources inside a VPC. NACLs are similar to Security Groups in a way that you can define incoming and outgoing traffic rules.

A few of the differences between NACLs and Security Groups:

NACLsSecurity Groups
Work on a Subnet levelAttached to network interfaces
Allow explicit Deny rules for ports and IP address rangesDefine only Allow rules for ports and IP address ranges
StatelessStateful

Create Network Access Control List

To create a Network Access Control List, you need to use the create_network_acl() method from the Boto3 library. In the example below, we are using the Boto3 EC2 resource to create the Network ACL:

import logging
import boto3
from botocore.exceptions import ClientError
import json
AWS_REGION = 'us-east-2'
# logger config
logger = logging.getLogger()
logging.basicConfig(level=logging.INFO,
                    format='%(asctime)s: %(levelname)s: %(message)s')
vpc_resource = boto3.resource("ec2", region_name=AWS_REGION)

def create_network_acl(vpc_id):
    """
    Creates a network access control list with the specified configuration.
    """
    try:
        response = vpc_resource.create_network_acl(
            VpcId=vpc_id,
            TagSpecifications=[
                {
                    'ResourceType':
                    'network-acl',
                    'Tags': [
                        {
                            'Key': 'Name',
                            'Value': 'hands-on-cloud-network-acl'
                        },
                    ]
                },
            ])
    except ClientError:
        logger.exception('Could not create a network access control list.')
        raise
    else:
        return response

if __name__ == '__main__':
    # Constants
    VPC_ID = 'vpc-048604f523ad01d74'
    logger.info(f'Creating a network access control list...')
    network_acl = create_network_acl(VPC_ID)
    logger.info(
        f'Network access control list created with ID: {network_acl.id}')

The required attribute is:

  • VpcId: Specifies the ID of the VPC.

The create_network_acl() method returns the NetworkAcl resource object containing the Network ACL ID on a successful request.

Here’s a code execution output:

Create Network ACL
Create Network ACL

Manage Network Access Control List Entries

Network ACL entries define the network traffic parameters to control the traffic on the ports and protocol level.

Here are the attributes of a network ACL rule:

  • Rule number. Rules are evaluated on a priority basis starting with the lowest to the highest numbered rule.
  • Type. The type of traffic, for example, HTTPS.
  • Protocol. You can specify all standard protocols to control the traffic flow.
  • Port range. Specifies the listening port or port range for incoming or outgoing traffic. For example, 22 for SSH traffic.
  • Source. The source CIDR range of the traffic. This applies to Inbound rules only.
  • Destination. The destination CIDR range for the traffic. This applies to Outbound rules only.
  • Allow/Deny. Specifies allow or deny conditions for the specified traffic.

Create Network Access Control List entry

To create a Network Access Control List entry, you need to use the create_network_acl_entry() method from the Boto3 library.

import logging
import boto3
from botocore.exceptions import ClientError
import json
AWS_REGION = 'us-east-2'
# logger config
logger = logging.getLogger()
logging.basicConfig(level=logging.INFO,
                    format='%(asctime)s: %(levelname)s: %(message)s')
vpc_client = boto3.client("ec2", region_name=AWS_REGION)

def create_network_acl_entry(cidr, nacl_id, from_port, to_port, protocol,
                             rule_action, rule_number):
    """
    Creates a network acl entry with the specified configuration.
    """
    try:
        response = vpc_client.create_network_acl_entry(CidrBlock=cidr,
                                                       Egress=False,
                                                       NetworkAclId=nacl_id,
                                                       PortRange={
                                                           'From': from_port,
                                                           'To': to_port,
                                                       },
                                                       Protocol=protocol,
                                                       RuleAction=rule_action,
                                                       RuleNumber=rule_number)
    except ClientError:
        logger.exception('Could not create a network acl entry.')
        raise
    else:
        return response

if __name__ == '__main__':
    # Constants
    CIDR = '0.0.0.0/0'
    NACL_ID = 'acl-0fc095d09c6319831'
    FROM_PORT = 22
    TO_PORT = 22
    PROTOCOL = '6'
    RULE_ACTION = 'allow'
    RULE_NUMBER = 101
    logger.info('Creating a network acl entry...')
    network_acl = create_network_acl_entry(CIDR, NACL_ID, FROM_PORT,
                                              TO_PORT, PROTOCOL, RULE_ACTION,
                                              RULE_NUMBER)
    logger.info('Network Acl Entry created.')

The required attributes are:

  • Egress: Specifies whether this is an egress rule (applied to traffic leaving the subnet).
  • NetworkAclId: The ID of the network ACL.
  • Protocol: The protocol number. 
ProtocolProtocol Number
TCP6
UDP17
ICMP1
All Protocols-1
Protocol Numbers
  • RuleAction:  Specifies whether to allow or deny the traffic that matches the rule.
  • RuleNumber: The rule number for the entry (for example, 100).

Optional arguments used in the above example:

  • CidrBlock: The IPv4 CIDR network range to allow or deny (for example, 192.128.0.0/24 ).
  • PortRange: TCP or UDP protocols: The range of ports the rule applies to. It is required if specifying protocol 6 (TCP) or 17 (UDP).
    • From: The first port in the range.
    • To: The last port in the range.

The create_network_acl_entry() method does not return any object on a successful request.

Here’s a code execution output:

Create Network ACL Entry
Create Network ACL Entry

Replace Network Access Control List entry

To replace an existing Network Access Control List entry, you need to use the replace_network_acl_entry() method from the Boto3 library.

import logging
import boto3
from botocore.exceptions import ClientError
import json
AWS_REGION = 'us-east-2'
# logger config
logger = logging.getLogger()
logging.basicConfig(level=logging.INFO,
                    format='%(asctime)s: %(levelname)s: %(message)s')
vpc_client = boto3.client("ec2", region_name=AWS_REGION)

def replace_network_acl_entry(cidr, nacl_id, from_port, to_port, protocol,
                              rule_action, rule_number):
    """
    Replaces a network acl entry with the specified configuration.
    """
    try:
        response = vpc_client.replace_network_acl_entry(CidrBlock=cidr,
                                                        Egress=False,
                                                        NetworkAclId=nacl_id,
                                                        PortRange={
                                                            'From': from_port,
                                                            'To': to_port,
                                                        },
                                                        Protocol=protocol,
                                                        RuleAction=rule_action,
                                                        RuleNumber=rule_number)
    except ClientError:
        logger.exception('Could not replace a network acl entry.')
        raise
    else:
        return response

if __name__ == '__main__':
    # Constants
    CIDR = '0.0.0.0/0'
    NACL_ID = 'acl-0fc095d09c6319831'
    FROM_PORT = 80
    TO_PORT = 80
    PROTOCOL = '6'
    RULE_ACTION = 'allow'
    RULE_NUMBER = 101
    logger.info('Replacing a network acl entry...')
    network_acl = replace_network_acl_entry(CIDR, NACL_ID, FROM_PORT,
                                               TO_PORT, PROTOCOL, RULE_ACTION,
                                               RULE_NUMBER)
    logger.info('Network Acl Entry replaced.')

The required attributes are:

  • Egress: Specifies whether this is an egress rule (applied to traffic leaving the subnet).
  • NetworkAclId: The ID of the network ACL.
  • Protocol: The protocol number.
ProtocolProtocol Number
TCP6
UDP17
ICMP1
All Protocols-1
Protocol Numbers
  • RuleAction:  Specifies whether to allow or deny the traffic that matches the rule.
  • RuleNumber: The rule number for the entry (for example, 100).

Optional arguments used in the above example:

  • CidrBlock: The IPv4 CIDR network range to allow or deny (for example, 192.128.0.0/24 ).
  • PortRange: TCP or UDP protocols: The range of ports the rule applies to. It is required if specifying protocol 6 (TCP) or 17 (UDP).
    • From: The first port in the range.
    • To: The last port in the range.

In the above example, we replace the SSH 22 port entry created in the previous example with the HTTP 80 port.

The replace_network_acl_entry() method does not return any object on a successful request.

Here’s a code execution output:

Replace Network ACL Entry
Replace Network ACL Entry

Delete Network Access Control List Entry

To delete an existing Network Access Control List entry, you need to use the delete_network_acl_entry() method from the Boto3 library.

import logging
import boto3
from botocore.exceptions import ClientError
import json
AWS_REGION = 'us-east-2'
# logger config
logger = logging.getLogger()
logging.basicConfig(level=logging.INFO,
                    format='%(asctime)s: %(levelname)s: %(message)s')
vpc_client = boto3.client("ec2", region_name=AWS_REGION)

def delete_network_acl_entry(nacl_id, rule_number):
    """
    Deletes the specified network acl entry.
    """
    try:
        response = vpc_client.delete_network_acl_entry(Egress=False,
                                                       NetworkAclId=nacl_id,
                                                       RuleNumber=rule_number)
    except ClientError:
        logger.exception('Could not delete the network acl entry.')
        raise
    else:
        return response

if __name__ == '__main__':
    # Constants
    NACL_ID = 'acl-0fc095d09c6319831'
    RULE_NUMBER = 101
    network_acl = delete_network_acl_entry(NACL_ID, RULE_NUMBER)
    logger.info('Network acl entry deleted successfully.')

The required attributes are:

  • Egress: Specifies whether this is an egress rule (applied to traffic leaving the subnet).
  • NetworkAclId: The ID of the network ACL.
  • RuleNumber: The rule number for the entry (for example, 100).

In the above example, we delete the HTTP 80 port entry with Rule Number 101 created in the previous example.

The delete_network_acl_entry() method does not return any object on a successful request.

Here’s a code execution output:

Delete Network ACL Entry
Delete Network ACL Entry

Describe Network Access Control Lists

To get the current Network ACLs information in your AWS account, you need to use the describe_network_acls() method from the Boto3 library.

This method allows you to describe existing Network ACLs using two methods, Network ACL Ids, and Filters. We will use Filters in the below example.

We will use the Boto3 library paginator object to get the complete output from the describe_network_acls() method.

import logging
import boto3
from botocore.exceptions import ClientError
import json
AWS_REGION = 'us-east-2'
# logger config
logger = logging.getLogger()
logging.basicConfig(level=logging.INFO,
                    format='%(asctime)s: %(levelname)s: %(message)s')
vpc_client = boto3.client("ec2", region_name=AWS_REGION)

def describe_network_acls(tag, tag_values, max_items):
    """
    Describes one or more of your network acls.
    """
    try:
        # creating paginator object for describe_network_acls() method
        paginator = vpc_client.get_paginator('describe_network_acls')
        # creating a PageIterator from the paginator
        response_iterator = paginator.paginate(
            Filters=[{
                'Name': f'tag:{tag}',
                'Values': tag_values
            }],
            PaginationConfig={'MaxItems': max_items})
        full_result = response_iterator.build_full_result()
        network_acls_list = []
        for page in full_result['NetworkAcls']:
            network_acls_list.append(page)
    except ClientError:
        logger.exception('Could not describe Network Acls.')
        raise
    else:
        return network_acls_list

if __name__ == '__main__':
    # Constants
    TAG = 'Name'
    TAG_VALUES = ['hands-on-cloud-network-acl']
    MAX_ITEMS = 10
    network_acls = describe_network_acls(TAG, TAG_VALUES, MAX_ITEMS)
    logger.info('Network Acls Details: ')
    for network_acl in network_acls:
        logger.info(json.dumps(network_acl, indent=4) + '\n')

Optional arguments used in the above example:

  • Filters: Specifies the different available filters for narrowing the Network ACLs search criteria. For all available filters, refer the official Boto3 documentation.
  • PaginationConfig: Specifies the configuration for the Paginator object. We have used the MaxItems attribute, which specifies the maximum number of Network ACLs to return in one request.

The describe_network_acls() method returns a python dictionary object as an output.

Here’s a code execution output:

Describe Network ACLs
Describe Network ACLs

Delete Network Access Control List

To delete an existing Network Access Control List, you need to use the delete_network_acl() method from the Boto3 library.

import logging
import boto3
from botocore.exceptions import ClientError
import json
AWS_REGION = 'us-east-2'
# logger config
logger = logging.getLogger()
logging.basicConfig(level=logging.INFO,
                    format='%(asctime)s: %(levelname)s: %(message)s')
vpc_client = boto3.client("ec2", region_name=AWS_REGION)

def delete_network_acl(network_acl_id):
    """
    Deletes the specified Network Acl.
    """
    try:
        response = vpc_client.delete_network_acl(NetworkAclId=network_acl_id)
    except ClientError:
        logger.exception('Could not delete the Network Acl.')
        raise
    else:
        return response

if __name__ == '__main__':
    # Constants
    NETWORK_ACL_ID = 'acl-0f01659015e478041'
    network_acl = delete_network_acl(NETWORK_ACL_ID)
    logger.info(f'Network Acl {NETWORK_ACL_ID} is deleted successfully.')

The required attribute is:

  • NetworkAclId: Specifies the ID of the network ACL to be deleted.

The delete_network_acl() method does not return any object on a successful request.

Here’s a code execution output:

Delete Network ACL
Delete Network ACL

How to manage Internet Gateways using Boto3?

An Internet Gateway (IGW) enables the communication between public resources in your AWS Virtual Private Cloud and the Internet. You can only have one IGW per VPC. There is no additional cost for having an Internet Gateway in your account.

Create Internet Gateway

To create an Internet gateway, you need to use the create_internet_gateway() method from the Boto3 library.

import logging
import boto3
from botocore.exceptions import ClientError
import json
AWS_REGION = 'us-east-2'
# logger config
logger = logging.getLogger()
logging.basicConfig(level=logging.INFO,
                    format='%(asctime)s: %(levelname)s: %(message)s')
vpc_client = boto3.client("ec2", region_name=AWS_REGION)

def create_igw():
    """
    Creates an internet gateway with the specified configuration.
    """
    try:
        response = vpc_client.create_internet_gateway(TagSpecifications=[
            {
                'ResourceType': 'internet-gateway',
                'Tags': [
                    {
                        'Key': 'Name',
                        'Value': 'hands-on-cloud-igw'
                    },
                ]
            },
        ])
    except ClientError:
        logger.exception('Could not create the internet gateway.')
        raise
    else:
        return response

if __name__ == '__main__':
    # Constants
    logger.info('Creating an internet gateway...')
    igw = create_igw()
    logger.info(f'Internet gateway created with: {json.dumps(igw, indent=4)}')

An optional argument used in the above example:

  • TagSpecifications: Specifies the resource tags for the Internet Gateway.

When you create an Internet Gateway, it is not attached to any VPC by default.

The create_internet_gateway() method returns a python dictionary object as an output.

Here’s the code execution output:

Create Internet Gateway
Create Internet Gateway

Attach Internet Gateway to a VPC

To attach an Internet Gateway to a VPC, you need to use the attach_internet_gateway() method from the Boto3 library.

import logging
import boto3
from botocore.exceptions import ClientError
import json
AWS_REGION = 'us-east-2'
# logger config
logger = logging.getLogger()
logging.basicConfig(level=logging.INFO,
                    format='%(asctime)s: %(levelname)s: %(message)s')
vpc_client = boto3.client("ec2", region_name=AWS_REGION)

def attach_igw_to_vpc(internet_gateway_id, vpc_id):
    """
    Attaches an internet gateway to a VPC.
    """
    try:
        response = vpc_client.attach_internet_gateway(
            InternetGatewayId=internet_gateway_id, VpcId=vpc_id)
    except ClientError:
        logger.exception('Could not attach an internet gateway to a VPC.')
        raise
    else:
        return response

if __name__ == '__main__':
    # Constants
    INTERNET_GATEWAY_ID = 'igw-011479e8fea5256a2'
    VPC_ID = 'vpc-09a7bc60ca95ceea2'
    logger.info('Attaching an internet gateway to the VPC...')
    igw = attach_igw_to_vpc(INTERNET_GATEWAY_ID, VPC_ID)
    logger.info(
        f'Internet gateway {INTERNET_GATEWAY_ID} attahced to the VPC {VPC_ID} successfully.'
    )

The required attributes are:

  • InternetGatewayId: Specifies the ID of the internet gateway.
  • VpcId: Specifies the ID of the VPC.

The attach_internet_gateway() method does not return any object on a successful request.

Here’s the code execution output:

Attach Internet Gateway
Attach Internet Gateway

Detach Internet Gateway from a VPC

To detach an Internet Gateway from a VPC, you need to use the detach_internet_gateway() method from the Boto3 library.

Before detaching the Internet Gateway, the specified VPC must not contain any running EC2 instances with Elastic IP addresses or public IPv4 addresses.

import logging
import boto3
from botocore.exceptions import ClientError
import json
AWS_REGION = 'us-east-2'
# logger config
logger = logging.getLogger()
logging.basicConfig(level=logging.INFO,
                    format='%(asctime)s: %(levelname)s: %(message)s')
vpc_client = boto3.client("ec2", region_name=AWS_REGION)

def detach_igw_to_vpc(internet_gateway_id, vpc_id):
    """
    Detaches an internet gateway from a VPC.
    """
    try:
        response = vpc_client.detach_internet_gateway(
            InternetGatewayId=internet_gateway_id, VpcId=vpc_id)
    except ClientError:
        logger.exception('Could not detach an internet gateway from a VPC.')
        raise
    else:
        return response

if __name__ == '__main__':
    # Constants
    INTERNET_GATEWAY_ID = 'igw-011479e8fea5256a2'
    VPC_ID = 'vpc-09a7bc60ca95ceea2'
    logger.info('Detaching an internet gateway from the VPC...')
    igw = detach_igw_to_vpc(INTERNET_GATEWAY_ID, VPC_ID)
    logger.info(
        f'Internet gateway {INTERNET_GATEWAY_ID} detahced from the VPC {VPC_ID} successfully.'
    )

The required attributes are:

  • InternetGatewayId: Specifies the ID of the internet gateway.
  • VpcId: Specifies the ID of the VPC.

The detach_internet_gateway() method does not return any object on a successful request.

Here’s the code execution output:

Detach Internet Gateway
Detach Internet Gateway

Describe Internet Gateways

To get detailed information of the existing Internet Gateways in your AWS account, you need to use the describe_internet_gateways() method from the Boto3 library. 

This method allows you to describe existing Internet Gateways by using Internet Gateway Ids and Filters. We will use Filters in the example below.

We will use the Boto3 library paginator object to get the complete output from the describe_internet_gateways() method.

import logging
import boto3
from botocore.exceptions import ClientError
import json
AWS_REGION = 'us-east-2'
# logger config
logger = logging.getLogger()
logging.basicConfig(level=logging.INFO,
                    format='%(asctime)s: %(levelname)s: %(message)s')
vpc_client = boto3.client("ec2", region_name=AWS_REGION)

def describe_internet_gateways(tag, tag_values, max_items):
    """
    Describes one or more of your internet gateways.
    """
    try:
        # creating paginator object for describe_internet_gateways() method
        paginator = vpc_client.get_paginator('describe_internet_gateways')
        # creating a PageIterator from the paginator
        response_iterator = paginator.paginate(
            Filters=[{
                'Name': f'tag:{tag}',
                'Values': tag_values
            }],
            PaginationConfig={'MaxItems': max_items})
        full_result = response_iterator.build_full_result()
        internet_gateways_list = []
        for page in full_result['InternetGateways']:
            internet_gateways_list.append(page)
    except ClientError:
        logger.exception('Could not describe Internet Gateways.')
        raise
    else:
        return internet_gateways_list

if __name__ == '__main__':
    # Constants
    TAG = 'Name'
    TAG_VALUES = ['hands-on-cloud-igw']
    MAX_ITEMS = 10
    internet_gateways = describe_internet_gateways(TAG, TAG_VALUES, MAX_ITEMS)
    logger.info('Internet Gateways Details: ')
    for internet_gateway in internet_gateways:
        logger.info(json.dumps(internet_gateway, indent=4) + '\n')

Optional arguments used in the above example:

  • Filters: Specifies the different available filters for narrowing the Internet Gateways search criteria. For all available filters, refer here.
  • PaginationConfig: Specifies the configuration for the Paginator object. We have used the MaxItems attribute, which specifies the maximum number of Internet Gateways to return in one request.

The describe_internet_gateways() method returns a python dictionary object as an output.

Here’s a code execution output:

Describe Internet Gateways
Describe Internet Gateways

Delete Internet Gateway

To delete an existing Internet Gateway, you need to use the delete_internet_gateway() method from the Boto3 library.

import logging
import boto3
from botocore.exceptions import ClientError
import json
AWS_REGION = 'us-east-2'
# logger config
logger = logging.getLogger()
logging.basicConfig(level=logging.INFO,
                    format='%(asctime)s: %(levelname)s: %(message)s')
vpc_client = boto3.client("ec2", region_name=AWS_REGION)

def delete_internet_gateway(internet_gateway_id):
    """
    Deletes the specified internet gateway.
    """
    try:
        response = vpc_client.delete_internet_gateway(
            InternetGatewayId=internet_gateway_id)
    except ClientError:
        logger.exception('Could not delete the Internet Gateway.')
        raise
    else:
        return response

if __name__ == '__main__':
    # Constants
    INTERNET_GATEWAY_ID = 'igw-011479e8fea5256a2'
    network_acl = delete_internet_gateway(INTERNET_GATEWAY_ID)
    logger.info(
        f'Internet Gateway {INTERNET_GATEWAY_ID} is deleted successfully.')

The required attribute is:

  • InternetGatewayId: Specifies the ID of the Internet Gateway to be deleted.

The delete_internet_gateway() method does not return any object on a successful request.

Here’s a code execution output:

Delete Internet Gateway
Delete Internet Gateway

How to manage NAT Gateways using Boto3?

A NAT (Network Address Translation) Gateway allows VPC resources (for example, EC2 instances) launched in a private subnet to have internet connectivity without exposing them to the Internet.

NAT gateway routes traffic from your private subnet to the Internet or other AWS services. For outbound traffic, the source IP address is replaced with the NAT gateway’s address. 

Similarly, the NAT gateway translates the address back to those instance’s private IP addresses for inbound traffic.

Create NAT Gateway

To create a NAT gateway, you need to use the create_nat_gateway() method from the Boto3 library.

Below are few requirements before creating a NAT gateway in a VPC:

  • VPC must have an Internet Gateway attached to it before creating a NAT Gateway in that VPC.
  • NAT Gateway requires an Elastic IPv4 IP address before NAT Gateway is created.

In addition, you have to use the EC2.Waiter.NatGatewayAvailable class to ensure the NAT Gateway is created and available to use. That’s a correct way to wait for the successful completion of the NAT Gateway creation operation.

import logging
import boto3
from datetime import date, datetime
from botocore.exceptions import ClientError
import json
AWS_REGION = 'us-east-2'
# logger config
logger = logging.getLogger()
logging.basicConfig(level=logging.INFO,
                    format='%(asctime)s: %(levelname)s: %(message)s')
vpc_client = boto3.client("ec2", region_name=AWS_REGION)

def json_datetime_serializer(obj):
    """
    Helper method to serialize datetime fields
    """
    if isinstance(obj, (datetime, date)):
        return obj.isoformat()
    raise TypeError("Type %s not serializable" % type(obj))

def wait_nat_creation(nat_gateway_id):
    """
    Check if successful state is reached every 15 seconds until a successful state is reached.
    An error is returned after 40 failed checks.
    """
    try:
        waiter = vpc_client.get_waiter('nat_gateway_available')
        waiter.wait(NatGatewayIds=[nat_gateway_id])
    except ClientError:
        logger.exception(f'Could not create the NAT gateway.')
        raise

def allocate_address():
    """
    Allocates an Elastic IP address to use with an NAT Gateway in a VPC.
    """
    try:
        response = vpc_client.allocate_address(Domain='vpc')
    except ClientError:
        logger.exception(f'Could not create the NAT gateway.')
        raise
    else:
        return response['AllocationId']

def create_nat(subnet_id):
    """
    Creates a NAT gateway in the specified subnet.
    """
    try:
        # allocate IPV4 address for NAT gateway
        public_ip_allocation_id = allocate_address()
        # create NAT gateway
        response = vpc_client.create_nat_gateway(
            AllocationId=public_ip_allocation_id,
            SubnetId=subnet_id,
            TagSpecifications=[{
                'ResourceType':
                'natgateway',
                'Tags': [{
                    'Key': 'Name',
                    'Value': 'hands-on-cloud-nat-gateway'
                }]
            }])
        nat_gateway_id = response['NatGateway']['NatGatewayId']
        # wait until the NAT gateway is available
        wait_nat_creation(nat_gateway_id)
    except ClientError:
        logger.exception(f'Could not create the NAT gateway.')
        raise
    else:
        return response

if __name__ == '__main__':
    # Constants
    SUBNET_ID = 'subnet-0b6498751bd99fce7'
    logger.info(f'Creating a NAT gateway...')
    nat = create_nat(SUBNET_ID)
    logger.info(
        f'NAT gateway created with: {json.dumps(nat, indent=4, default=json_datetime_serializer)}'
    )

The required attributes are:

  • AllocationId: Specifies the allocation ID of an Elastic IP address to associate with the NAT gateway.
  • SubnetId: Specifies the subnet in which to create the NAT gateway.

An optional argument used in the above example:

  • TagSpecifications: Specifies the resource tags for the NAT Gateway.

In the example above, we’re using additional json_datetime_serializer() method to serialize (convert to string) datetime.datetime fields returned by the create_nat_gateway() method.

The create_nat_gateway() method returns a python dictionary object as an output.

Here’s the code execution output:

Create NAT Gateway
Create NAT Gateway

Describe NAT Gateways

To get the details of the existing NAT Gateways in your AWS account, you need to use the describe_nat_gateways() method from the Boto3 library. 

This method allows you to describe existing NAT Gateways by using NAT Gateway Ids and Filters. We will use Filters in the example below.

We will use the Boto3 library paginator object to get the complete output from the describe_nat_gateways() method.

import logging
import boto3
from datetime import date, datetime
from botocore.exceptions import ClientError
import json
AWS_REGION = 'us-east-2'
# logger config
logger = logging.getLogger()
logging.basicConfig(level=logging.INFO,
                    format='%(asctime)s: %(levelname)s: %(message)s')
vpc_client = boto3.client("ec2", region_name=AWS_REGION)

def json_datetime_serializer(obj):
    """
    Helper method to serialize datetime fields
    """
    if isinstance(obj, (datetime, date)):
        return obj.isoformat()
    raise TypeError("Type %s not serializable" % type(obj))

def describe_nat_gateways(tag, tag_values, max_items):
    """
    Describes one or more of your NAT gateways.
    """
    try:
        # creating paginator object for describe_nat_gateways() method
        paginator = vpc_client.get_paginator('describe_nat_gateways')
        # creating a PageIterator from the paginator
        response_iterator = paginator.paginate(
            Filters=[{
                'Name': f'tag:{tag}',
                'Values': tag_values
            }],
            PaginationConfig={'MaxItems': max_items})
        full_result = response_iterator.build_full_result()
        nat_gateways_list = []
        for page in full_result['NatGateways']:
            nat_gateways_list.append(page)
    except ClientError:
        logger.exception('Could not describe NAT Gateways.')
        raise
    else:
        return nat_gateways_list

if __name__ == '__main__':
    # Constants
    TAG = 'Name'
    TAG_VALUES = ['hands-on-cloud-nat-gateway']
    MAX_ITEMS = 10
    nat_gateways = describe_nat_gateways(TAG, TAG_VALUES, MAX_ITEMS)
    logger.info('NAT Gateways Details: ')
    for nat_gateway in nat_gateways:
        logger.info(
            json.dumps(nat_gateway, indent=4, default=json_datetime_serializer)
            + '\n')

Optional arguments used in the above example:

  • Filters: Specifies the different available filters for narrowing the NAT Gateways search criteria. For all available filters, refer here.
  • PaginationConfig: Specifies the configuration for the Paginator object. We have used the MaxItems attribute, which specifies the maximum number of NAT Gateways to return in one request.

In the example above, we’re using additional json_datetime_serializer() method to serialize (convert to string) datetime.datetime fields returned by the describe_nat_gateways() method.

The describe_nat_gateways() method returns a python dictionary object as an output.

Here’s a code execution output:

Describe NAT Gateways
Describe NAT Gateways

Delete NAT Gateway

To delete an existing NAT Gateway, you need to use the delete_nat_gateway() method from the Boto3 library.

Deleting a public NAT gateway disassociates its Elastic IP address but does not release the Elastic IP address from your account.

import logging
import boto3
from botocore.exceptions import ClientError
import json
AWS_REGION = 'us-east-2'
# logger config
logger = logging.getLogger()
logging.basicConfig(level=logging.INFO,
                    format='%(asctime)s: %(levelname)s: %(message)s')
vpc_client = boto3.client("ec2", region_name=AWS_REGION)

def delete_nat_gateway(nat_gateway_id):
    """
    Deletes the specified NAT gateway.
    """
    try:
        response = vpc_client.delete_nat_gateway(NatGatewayId=nat_gateway_id)
    except ClientError:
        logger.exception('Could not delete the NAT Gateway.')
        raise
    else:
        return response

if __name__ == '__main__':
    # Constants
    NAT_GATEWAY_ID = 'nat-0dce41a56b339d240'
    nat_gateway = delete_nat_gateway(NAT_GATEWAY_ID)
    logger.info(f'NAT Gateway {NAT_GATEWAY_ID} is deleted successfully.')

The required attribute is:

  • NatGatewayId: Specifies the ID of the NAT Gateway to be deleted.

The delete_nat_gateway() method does not return any object on a successful request.

Here’s a code execution output:

Delete NAT Gateway
Delete NAT Gateway

How to manage Route Tables using Boto3?

Each subnet in a VPC must be associated with a routing table, which controls the network traffic routing for the subnet.

You can explicitly associate a subnet with a custom route table. Otherwise, the subnet is implicitly associated with the default route table.

Only one route table can be associated with a subnet at a time. However, you can associate multiple subnets with the same route table.

Create Route Table

To create a Route table, you need to use the create_route_table() method from the Boto3 library.

import logging
import boto3
from botocore.exceptions import ClientError
import json
AWS_REGION = 'us-east-2'
# logger config
logger = logging.getLogger()
logging.basicConfig(level=logging.INFO,
                    format='%(asctime)s: %(levelname)s: %(message)s')
vpc_client = boto3.client("ec2", region_name=AWS_REGION)

def create_rt(vpc_id):
    """
    Creates a route table for the specified VPC with the specified configuration.
    """
    try:
        response = vpc_client.create_route_table(
            VpcId=vpc_id,
            TagSpecifications=[
                {
                    'ResourceType': 'route-table',
                    'Tags': [
                        {
                            'Key': 'Name',
                            'Value': 'hands-on-cloud-rt'
                        },
                    ]
                },
            ])
    except ClientError:
        logger.exception('Could not create the route table.')
        raise
    else:
        return response

if __name__ == '__main__':
    # Constants
    VPC_ID = 'vpc-048604f523ad01d74'
    logger.info(f'Creating a route ta ble...')
    rt = create_rt(VPC_ID)
    logger.info(f'Route table created with: {json.dumps(rt, indent=4)}')

The required attribute is:

  • VpcId: Specifies the ID of the VPC.

An optional argument used in the above example:

  • TagSpecifications: Specifies the resource tags for the Route table.

When you create a Route table, it is not attached to any subnet by default. 

The create_route_table() method returns a python dictionary object as an output.

Here’s the code execution output:

Create Route table
Create Route table

Create Route

To create a route in a routing table, you need to use the create_route() method from the Boto3 library.

import logging
import boto3
from botocore.exceptions import ClientError
import json
AWS_REGION = 'us-east-2'
# logger config
logger = logging.getLogger()
logging.basicConfig(level=logging.INFO,
                    format='%(asctime)s: %(levelname)s: %(message)s')
vpc_client = boto3.client("ec2", region_name=AWS_REGION)

def create_route(destination_cidr_block, gateway_id, route_table_id):
    """
    Creates a route in a route table within a VPC.
    """
    try:
        response = vpc_client.create_route(
            DestinationCidrBlock=destination_cidr_block,
            GatewayId=gateway_id,
            RouteTableId=route_table_id)
    except ClientError:
        logger.exception('Could not create the route.')
        raise
    else:
        return response

if __name__ == '__main__':
    # Constants
    DESTINATION_CIDR_BLOCK = '0.0.0.0/0'
    GATEWAY_ID = 'igw-005101a6dcc5d1e4f'
    ROUTE_TABLE_ID = 'rtb-04c3a19f68e99ac19'
    logger.info('Adding a route...')
    route = create_route(DESTINATION_CIDR_BLOCK, GATEWAY_ID, ROUTE_TABLE_ID)
    logger.info(
        f'Route added to the route table: {json.dumps(route, indent=4)}')

The required attribute is:

  • RouteTableId: Specifies the ID of the Route table.

Optional arguments used in the above example:

  • GatewayId: Specifies the ID of an internet gateway attached to the VPC.
  • DestinationCidrBlock: The IPv4 CIDR address range used for the destination match.

In the above example, we create a route for the routing table. The route matches all traffic (0.0.0.0/0) and routes it to the specified Internet gateway.

The create_route() method returns a python dictionary object as an output.

Here’s the code execution output:

Create Route
Create Route

Replace Route

To replace a route in a route table, you need to use the replace_route() method from the Boto3 library.

import logging
import boto3
from botocore.exceptions import ClientError
import json
AWS_REGION = 'us-east-2'
# logger config
logger = logging.getLogger()
logging.basicConfig(level=logging.INFO,
                    format='%(asctime)s: %(levelname)s: %(message)s')
vpc_client = boto3.client("ec2", region_name=AWS_REGION)

def replace_route(destination_cidr_block, gateway_id, route_table_id):
    """
    Replaces a route in a route table within a VPC.
    """
    try:
        response = vpc_client.replace_route(
            DestinationCidrBlock=destination_cidr_block,
            NatGatewayId=gateway_id,
            RouteTableId=route_table_id)
    except ClientError:
        logger.exception('Could not replace the route.')
        raise
    else:
        return response

if __name__ == '__main__':
    # Constants
    DESTINATION_CIDR_BLOCK = '0.0.0.0/0'
    NAT_GATEWAY_ID = 'nat-0924940267273e639'
    ROUTE_TABLE_ID = 'rtb-04c3a19f68e99ac19'
    logger.info('Replacing a route...')
    route = replace_route(DESTINATION_CIDR_BLOCK, NAT_GATEWAY_ID,
                          ROUTE_TABLE_ID)
    logger.info('Route replaced.')

The required attribute is:

  • RouteTableId: Specifies the ID of the Route table.

Optional arguments used in the above example:

  • NatGatewayId: Specifies the ID of a NAT gateway attached to the VPC.
  • DestinationCidrBlock: The IPv4 CIDR address range used for the destination match.

In the above example, we replace a route created in the last example and route the traffic to the NAT gateway instead of the Internet Gateway. The route matches all traffic (0.0.0.0/0) and routes it to the specified NAT gateway.

The replace_route() method does not return any object on a successful request.

Here’s a code execution output:

Replace Route
Replace Route

Delete Route

To delete an existing route from a Route table, you need to use the delete_route() method from the Boto3 library.

import logging
import boto3
from botocore.exceptions import ClientError
import json
AWS_REGION = 'us-east-2'
# logger config
logger = logging.getLogger()
logging.basicConfig(level=logging.INFO,
                    format='%(asctime)s: %(levelname)s: %(message)s')
vpc_client = boto3.client("ec2", region_name=AWS_REGION)

def delete_route(destination_cidr_block, route_table_id):
    """
    Deletes the specified route from the specified route table.
    """
    try:
        response = vpc_client.delete_route(
            DestinationCidrBlock=destination_cidr_block,
            RouteTableId=route_table_id)
    except ClientError:
        logger.exception('Could not delete the route.')
        raise
    else:
        return response

if __name__ == '__main__':
    # Constants
    DESTINATION_CIDR_BLOCK = '0.0.0.0/0'
    ROUTE_TABLE_ID = 'rtb-04c3a19f68e99ac19'
    logger.info('Deleting a route...')
    route = delete_route(DESTINATION_CIDR_BLOCK, ROUTE_TABLE_ID)
    logger.info('Route deleted succussfully.')

The required attribute is:

  • RouteTableId: Specifies the ID of the Route table.

An optional argument used in the above example:

  • DestinationCidrBlock: The IPv4 CIDR address range used for the destination match.

The delete_route() method does not return any object on a successful request.

Here’s a code execution output:

Delete Route
Delete Route

Associate Route Table

To associate the specified route table with the specified subnet, you need to use the associate_route_table() method from the Boto3 library.

A route table can be associated with multiple subnets.

import logging
import boto3
from botocore.exceptions import ClientError
import json
AWS_REGION = 'us-east-2'
# logger config
logger = logging.getLogger()
logging.basicConfig(level=logging.INFO,
                    format='%(asctime)s: %(levelname)s: %(message)s')
vpc_client = boto3.client("ec2", region_name=AWS_REGION)

def associate_route_table(route_table_id, subnet_id):
    """
    Associates a subnet in the VPC with a route table.
    """
    try:
        response = vpc_client.associate_route_table(
            RouteTableId=route_table_id, SubnetId=subnet_id)
    except ClientError:
        logger.exception(
            'Could not associate the route table with the subnet.')
        raise
    else:
        return response

if __name__ == '__main__':
    # Constants
    SUBNET_ID = 'subnet-0b6498751bd99fce7'
    ROUTE_TABLE_ID = 'rtb-04c3a19f68e99ac19'
    logger.info('Associting a route table with the subnet...')
    route = associate_route_table(ROUTE_TABLE_ID, SUBNET_ID)
    logger.info(
        f'Route table associated succussfully: {json.dumps(route, indent=4)}')

The required attribute is:

  • RouteTableId: Specifies the ID of the Route table.

An optional argument used in the above example:

  • SubnetId: Specifies the ID of the Subnet.

The associate_route_table() method returns a python dictionary object as an output.

Here’s a code execution output:

Associate Route Table
Associate Route Table

Disassociate Route Table

To disassociate the specified route table from the specified subnet, you need to use the disassociate_route_table() method from the Boto3 library.

import logging
import boto3
from botocore.exceptions import ClientError
import json
AWS_REGION = 'us-east-2'
# logger config
logger = logging.getLogger()
logging.basicConfig(level=logging.INFO,
                    format='%(asctime)s: %(levelname)s: %(message)s')
vpc_client = boto3.client("ec2", region_name=AWS_REGION)

def disassociate_route_table(associatation_id):
    """
    Disassociates the route table from the subnet.
    """
    try:
        response = vpc_client.disassociate_route_table(
            AssociationId=associatation_id)
    except ClientError:
        logger.exception(
            'Could not disassociate the route table from the subnet.')
        raise
    else:
        return response

if __name__ == '__main__':
    # Constants
    ASSOCIATATION_ID = 'rtbassoc-0f0bfb6e153acd6aa'
    logger.info('Disssociting a route table from the subnet...')
    route = disassociate_route_table(ASSOCIATATION_ID)
    logger.info('Route table disassociated succussfully.')

The required attribute is:

  • AssociationId: Specifies the association ID representing the current association between the route table and subnet.

The disassociate_route_table() method does not return any object on a successful request.

Here’s a code execution output:

Disassociate Route table
Disassociate Route table

Describe Route Tables

To get the current Route tables information in your AWS account, you need to use the describe_route_tables() method from the Boto3 library.

This method allows you to describe existing Route tables using two methods, Route table Ids, and Filters. We will use Filters in the below example.

We will use the Boto3 library paginator object to get the complete output from the describe_route_tables() method.

import logging
import boto3
from botocore.exceptions import ClientError
import json
AWS_REGION = 'us-east-2'
# logger config
logger = logging.getLogger()
logging.basicConfig(level=logging.INFO,
                    format='%(asctime)s: %(levelname)s: %(message)s')
vpc_client = boto3.client("ec2", region_name=AWS_REGION)

def describe_route_tables(tag, tag_values, max_items):
    """
    Describes one or more of your Route tables.
    """
    try:
        # creating paginator object for describe_route_tables() method
        paginator = vpc_client.get_paginator('describe_route_tables')
        # creating a PageIterator from the paginator
        response_iterator = paginator.paginate(
            Filters=[{
                'Name': f'tag:{tag}',
                'Values': tag_values
            }],
            PaginationConfig={'MaxItems': max_items})
        full_result = response_iterator.build_full_result()
        route_tables_list = []
        for page in full_result['RouteTables']:
            route_tables_list.append(page)
    except ClientError:
        logger.exception('Could not describe Route tables.')
        raise
    else:
        return route_tables_list

if __name__ == '__main__':
    # Constants
    TAG = 'Name'
    TAG_VALUES = ['hands-on-cloud-rt']
    MAX_ITEMS = 10
    route_tables = describe_route_tables(TAG, TAG_VALUES, MAX_ITEMS)
    logger.info('Route tables Details: ')
    for route_table in route_tables:
        logger.info(json.dumps(route_table, indent=4) + '\n')

Optional arguments used in the above example:

  • Filters: Specifies the different available filters for narrowing the Route tables search criteria. For all available filters, refer here.
  • PaginationConfig: Specifies the configuration for the Paginator object. We have used the MaxItems attribute, which specifies the maximum number of Route tables to return in one request.

The describe_route_tables() method returns a python dictionary object as an output.

Here’s a code execution output:

Describe Route tables
Describe Route tables

Delete Route Table

To delete an existing route table, you need to use the delete_route_table() method from the Boto3 library.

import logging
import boto3
from botocore.exceptions import ClientError
import json
AWS_REGION = 'us-east-2'
# logger config
logger = logging.getLogger()
logging.basicConfig(level=logging.INFO,
                    format='%(asctime)s: %(levelname)s: %(message)s')
vpc_client = boto3.client("ec2", region_name=AWS_REGION)

def delete_route_table(route_table_id):
    """
    Deletes the specified route table.
    """
    try:
        response = vpc_client.delete_route_table(RouteTableId=route_table_id)
    except ClientError:
        logger.exception('Could not delete the route table.')
        raise
    else:
        return response

if __name__ == '__main__':
    # Constants
    ROUTE_TABLE_ID = 'rtb-04c3a19f68e99ac19'
    logger.info('Deleting a route table...')
    route = delete_route_table(ROUTE_TABLE_ID)
    logger.info('Route table deleted succussfully.')

The required attribute is:

  • RouteTableId: Specifies the ID of the Route table.

The delete_route_table() method does not return any object on a successful request.

Here’s a code execution output:

Delete Route table
Delete Route table

Summary

This article covered how to use the Python Boto3 library to programmatically interact with Amazon Virtual Private Cloud (Amazon VPC) service and create, manage, and perform management activities for AWS VPC, Subnets, Security Groups, Network ACLs, Internet Gateways, NAT Gateways, and Route Tables.