Transit Gateway is a service provided by AWS that allows you to create a central hub for your network traffic, making it easier to connect your VPCs, on-premises data centres, and remote networks. This blog post will discuss a code snippet demonstrating how to use the AWS SDK (boto3) to check for overlapping CIDR blocks between a VPC and a Transit Gateway route table before accepting an attachment request.
The code snippet starts by importing the boto3 library, the official SDK for AWS, and the ipaddress library, which works with IP addresses and networks. It then defines a Python class called “TransitGateway” which contains a method “accept_tgw_attachment”.
import boto3
import ipaddressThe first step in the method is to create a client for the EC2 service. This is done using the boto3.client() method and passing in the service name ‘ec2’ as a parameter. The method then sets up a Transit Gateway attachment ID and a role name as class variables.
class TransitGateway:
def init(self):
# Create a client for the EC2 service
self.client = boto3.client('ec2')
# Variables
self.tgw_attachment_id = 'tgw-attach-abcdef123456'
self.ROLE_NAME = "TestAdminRole"
self.tgw_route_table_id = "tgw-rtb-abcdef123456"The next step is to use the EC2 client to describe the Transit Gateway attachment using the ID specified in the class variable. This is done using the describe_transit_gateway_vpc_attachments() method, which returns information about the attachment, such as the VPC ID and the account number of the VPC owner.
def accept_tgw_attachment(self):
try:
# Describe the attachments
response = self.client.describe_transit_gateway_vpc_attachments(TransitGatewayAttachmentIds=[self.tgw_attachment_id])
# Extract the VPC ID and Account Number
vpc_id = response['TransitGatewayVpcAttachments'][0]['VpcId']
vpc_account_number = response['TransitGatewayVpcAttachments'][0]['VpcOwnerId']Once the VPC ID and account number have been extracted, the code assumes a role in the account identified by the account number using the boto3 library’s STS (Security Token Service) client. The STS client’s assume_role() method is used to assume a role in another account and returns temporary credentials that can be used to access resources in the other account.
The temporary credentials are then used to create a new session in the other account using the boto3.Session() method. This session is then used to create a new EC2 client that can be used to access resources in the other account.
# Assume a role in the account identified by the account number
sts_client = boto3.client('sts')
role_arn = f'arn:aws:iam::{vpc_account_number}:role/{self.ROLE_NAME}'
assumed_role_object = sts_client.assume_role(RoleArn=role_arn, RoleSessionName="session1")
# Extract the temporary credentials for the assumed role
credentials = assumed_role_object['Credentials']
# Use the temporary credentials to create a new session in the other account
session = boto3.Session(aws_access_key_id=credentials['AccessKeyId'],
aws_secret_access_key=credentials['SecretAccessKey'], aws_session_token=credentials['SessionToken'])
vpc_client = session.client('ec2')The next step is to use the new session’s EC2 client to retrieve the VPC’s CIDR block and the active routes of the Transit Gateway route table. The describe_vpcs() method is used to retrieve the VPC’s CIDR block, and the search_transit_gateway_routes() method is used to retrieve the active routes of the Transit Gateway route table.
Once the VPC’s CIDR block and the routes’ CIDR blocks have been retrieved, the code converts them to IpNetwork objects using the ipaddress library. This allows for a more efficient comparison between the two sets of IP addresses.
# Retrieve the VPC CIDR block and routes
vpc_cidr = vpc_client.describe_vpcs(VpcIds=[vpc_id])['Vpcs'][0]['CidrBlock']
routes = self.client.search_transit_gateway_routes(TransitGatewayRouteTableId=self.tgw_route_table_id,Filters=[{'Name': 'state','Values': ['active']}])['Routes']
# Convert VPC cidr and routes cidr to IpNetwork
vpc_cidr_block = ipaddress.ip_network(vpc_cidr)
overlap = FalseThe final step is to iterate through the routes and check if there’s an overlap between the VPC’s CIDR block and any of the routes. If there’s an overlap, the code prints a message and aborts the attachment. If there’s no overlap, it prints a message and accepts the attachment request using the accept_transit_gateway_vpc_attachment() method.
# Check for overlap in the routes
for route in routes:
if 'DestinationCidrBlock' in route:
route_cidr_block = ipaddress.ip_network(route['DestinationCidrBlock'])
if vpc_cidr_block.overlaps(route_cidr_block):
overlap = True
break
# Accept transit gateway request if no overlap else print overlap message
if overlap:
print(f"VPC CIDR {vpc_cidr_block} overlaps with the routes in {self.tgw_route_table_id} route table, aborting attachment")
else:
print(f"VPC CIDR {vpc_cidr_block} block does not overlap with any of the routes in {self.tgw_route_table_id} route table")
# Accepting the attachment request
response = self.client.accept_transit_gateway_vpc_attachment(TransitGatewayAttachmentId=self.tgw_attachment_id)
http_status_code = response['ResponseMetadata']['HTTPStatusCode']
if http_status_code == 200:
print('Transit gateway attachment successful')
else:
print('There was a problem attaching transit gateway')
# Exception Handling
except KeyError as e:
print(f"Error: {e}")
print("Check the provided Transit Gateway Attachment Id, Route Table Id and/or Role Name")
except boto3.exceptions.Boto3Error as e:
print(f'Error: {e}')
except Exception as e:
print(f'Error: {e}')
# Creating object of TransitGateway
transit_gateway = TransitGateway()
transit_gateway.accept_tgw_attachment()This code snippet is a great example of how to use the boto3 library to automate the process of checking for overlapping CIDR blocks before accepting an attachment request. It can be easily integrated into your existing infrastructure and can save you a lot of time and effort by automating a task that would otherwise have to be done manually.
Additionally, the use of the ipaddress library makes the checking process more efficient and the use of temporary credentials and new session ensures the security of your network. This is an important aspect of your network security and it’s good to have this check in place to avoid any issues that may
Here’s the entire code snippet of the Transit Gateway Attachment
import boto3
import ipaddress
class TransitGateway:
def __init__(self):
# Create a client for the EC2 service
self.client = boto3.client('ec2')
# Variables
# self.tgw_attachment_id = 'tgw-attach-abcde123456'
self.ROLE_NAME = "TestAdminRole"
self.tgw_route_table_id = "tgw-rtb-abcde123456"
def accept_tgw_attachment(self):
try:
# Describe the attachments
response = self.client.describe_transit_gateway_vpc_attachments(TransitGatewayAttachmentIds=[self.tgw_attachment_id])
# Extract the VPC ID and Account Number
vpc_id = response['TransitGatewayVpcAttachments'][0]['VpcId']
vpc_account_number = response['TransitGatewayVpcAttachments'][0]['VpcOwnerId']
# Assume a role in the account identified by the account number
sts_client = boto3.client('sts')
role_arn = f'arn:aws:iam::{vpc_account_number}:role/{self.ROLE_NAME}'
assumed_role_object = sts_client.assume_role(RoleArn=role_arn, RoleSessionName="session1")
# Extract the temporary credentials for the assumed role
credentials = assumed_role_object['Credentials']
# Use the temporary credentials to create a new session in the other account
session = boto3.Session(aws_access_key_id=credentials['AccessKeyId'],
aws_secret_access_key=credentials['SecretAccessKey'],
aws_session_token=credentials['SessionToken'])
vpc_client = session.client('ec2')
# Retrieve the VPC CIDR block and routes
vpc_cidr = vpc_client.describe_vpcs(VpcIds=[vpc_id])['Vpcs'][0]['CidrBlock']
routes = self.client.search_transit_gateway_routes(TransitGatewayRouteTableId=self.tgw_route_table_id,
Filters=[{'Name': 'state','Values': ['active']}])['Routes']
# Convert VPC cidr and routes cidr to IpNetwork
vpc_cidr_block = ipaddress.ip_network(vpc_cidr)
overlap = False
# Check for overlap in the routes
for route in routes:
if 'DestinationCidrBlock' in route:
route_cidr_block = ipaddress.ip_network(route['DestinationCidrBlock'])
if vpc_cidr_block.overlaps(route_cidr_block):
overlap = True
break
# Accept transit gateway request if no overlap else print overlap message
if overlap:
print(f"VPC CIDR {vpc_cidr_block} overlaps with the routes in {self.tgw_route_table_id} route table, aborting attachment")
else:
print(f"VPC CIDR {vpc_cidr_block} block does not overlap with any of the routes in {self.tgw_route_table_id} route table")
# Accepting the attachment request
response = self.client.accept_transit_gateway_vpc_attachment(TransitGatewayAttachmentId=self.tgw_attachment_id)
http_status_code = response['ResponseMetadata']['HTTPStatusCode']
if http_status_code == 200:
print('Transit gateway attachment successful')
else:
print('There was a problem attaching transit gateway')
# Exception Handling
except KeyError as e:
print(f"Error: {e}")
print("Check the provided Transit Gateway Attachment Id, Route Table Id and/or Role Name")
except boto3.exceptions.Boto3Error as e:
print(f'Error: {e}')
except Exception as e:
print(f'Error: {e}')
# Creating object of TransitGateway
transit_gateway = TransitGateway()
transit_gateway.accept_tgw_attachment()
