AWS Transit Gateway CIDR Route Checker

check_tgs_cidr.py

#!/usr/bin/env python3
"""
AWS Transit Gateway CIDR Route Checker

This script checks if a specific CIDR block exists in an AWS Transit Gateway route table
and provides information about the associated account and attachment.

Requirements:
    - boto3 (latest version)
    - AWS CLI configured with appropriate profiles
    - AWS credentials with permissions for:
        - ec2:SearchTransitGatewayRoutes
        - ec2:DescribeTransitGatewayAttachments
        - organizations:DescribeAccount (for account name lookup)

Author: Abdallah Deeb <abdallah@deeb.me>
Version: 1.1
"""

import argparse
import boto3
import ipaddress
import sys
from botocore.exceptions import ClientError, NoCredentialsError, ProfileNotFound


def validate_cidr(cidr_block):
    """
    Validate that the provided string is a valid CIDR block.

    Args:
        cidr_block (str): The CIDR block to validate

    Returns:
        bool: True if valid CIDR block, False otherwise
    """
    try:
        ipaddress.ip_network(cidr_block, strict=False)
        return True
    except ValueError:
        return False


def check_cidr_and_get_account(tgw_route_table_id, cidr_block, network_profile='network', org_profile='root'):
    """
    Check if a CIDR block exists in a Transit Gateway route table and get associated account info.

    Args:
        tgw_route_table_id (str): The Transit Gateway Route Table ID
        cidr_block (str): The CIDR block to search for (e.g., '10.0.0.0/16')
        network_profile (str): AWS profile for EC2 operations (default: 'network')
        org_profile (str): AWS profile for Organizations operations (default: 'root')

    Returns:
        dict: Dictionary containing route information or None if not found
    """
    # Validate CIDR block format
    if not validate_cidr(cidr_block):
        print(f"Error: '{cidr_block}' is not a valid CIDR block format.")
        return None

    try:
        # Create session with the specified profile
        session = boto3.Session(profile_name=network_profile)
        ec2_client = session.client('ec2')
    except ProfileNotFound:
        print(f"Error: AWS profile '{network_profile}' not found. Please check your AWS configuration.")
        return None
    except NoCredentialsError:
        print("Error: No AWS credentials found. Please configure your AWS credentials.")
        return None

    try:
        response = ec2_client.search_transit_gateway_routes(
            TransitGatewayRouteTableId=tgw_route_table_id,
            Filters=[
                {
                    'Name': 'state',
                    'Values': ['active']
                },
                {
                    'Name': 'route-search.exact-match',
                    'Values': [cidr_block]
                }
            ]
        )

        found_routes = response.get('Routes', [])

        if not found_routes:
            print(f"✗ CIDR block '{cidr_block}' not found in route table '{tgw_route_table_id}'.")
            return None

        route_info = []

        for route in found_routes:
            route_data = {
                'cidr_block': cidr_block,
                'route_table_id': tgw_route_table_id,
                'destination_cidr': route.get('DestinationCidrBlock'),
                'state': route.get('State'),
                'type': route.get('Type')
            }

            attachments = route.get('TransitGatewayAttachments', [])
            if not attachments:
                print(f"⚠ Found route for '{cidr_block}' but no attachment ID was associated with it.")
                route_data['status'] = 'no_attachment'
                route_info.append(route_data)
                continue

            attachment_id = attachments[0].get('TransitGatewayAttachmentId')
            route_data['attachment_id'] = attachment_id

            if attachment_id:
                try:
                    attachment_response = ec2_client.describe_transit_gateway_attachments(
                        TransitGatewayAttachmentIds=[attachment_id]
                    )
                    attachment_details = attachment_response.get('TransitGatewayAttachments', [])

                    if attachment_details:
                        attachment = attachment_details[0]
                        account_id = attachment.get('ResourceOwnerId')
                        resource_type = attachment.get('ResourceType')
                        resource_id = attachment.get('ResourceId')

                        route_data.update({
                            'account_id': account_id,
                            'resource_type': resource_type,
                            'resource_id': resource_id,
                            'attachment_state': attachment.get('State')
                        })

                        print(f"✓ CIDR block '{cidr_block}' found in route table '{tgw_route_table_id}'")
                        print(f"  └─ Attachment ID: {attachment_id}")
                        print(f"  └─ Resource Type: {resource_type}")
                        print(f"  └─ Resource ID: {resource_id}")
                        print(f"  └─ Account ID: {account_id}")

                        # Try to get account name from Organizations
                        try:
                            org_session = boto3.Session(profile_name=org_profile)
                            org_client = org_session.client('organizations')
                            account_details = org_client.describe_account(AccountId=account_id)
                            account_name = account_details['Account']['Name']
                            route_data['account_name'] = account_name
                            print(f"  └─ Account Name: {account_name}")
                        except ProfileNotFound:
                            print(f"  └─ Account Name: Unable to retrieve (profile '{org_profile}' not found)")
                        except ClientError as org_e:
                            error_code = org_e.response['Error']['Code']
                            if error_code == 'AccessDenied':
                                print(f"  └─ Account Name: Access denied to Organizations service")
                            elif error_code == 'AccountNotFound':
                                print(f"  └─ Account Name: Account {account_id} not found in organization")
                            else:
                                print(f"  └─ Account Name: Unable to retrieve ({error_code})")
                        except Exception as org_e:
                            print(f"  └─ Account Name: Unable to retrieve ({str(org_e)})")

                    else:
                        print(f"⚠ Found route for '{cidr_block}' but could not retrieve details for attachment '{attachment_id}'.")
                        route_data['status'] = 'attachment_details_unavailable'

                except ClientError as e:
                    error_code = e.response['Error']['Code']
                    if error_code == 'InvalidTransitGatewayAttachmentID.NotFound':
                        print(f"⚠ Attachment '{attachment_id}' not found.")
                    else:
                        print(f"⚠ Error retrieving attachment details: {error_code}")
                    route_data['status'] = f'error_{error_code}'

            route_info.append(route_data)

        return route_info

    except ClientError as e:
        error_code = e.response['Error']['Code']
        if error_code == 'InvalidRouteTableID.NotFound':
            print(f"Error: Transit Gateway Route Table '{tgw_route_table_id}' not found.")
        elif error_code == 'UnauthorizedOperation':
            print(f"Error: Insufficient permissions to access Transit Gateway route table.")
        else:
            print(f"AWS Error: {error_code} - {e.response['Error']['Message']}")
        return None
    except Exception as e:
        print(f"Unexpected error occurred: {e}")
        return None


def main():
    """Main function to handle command line arguments and execute the script."""
    parser = argparse.ArgumentParser(
        description='Check if a CIDR block exists in an AWS Transit Gateway route table',
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Examples:
  %(prog)s 10.0.0.0/16
  %(prog)s 192.168.1.0/24 --route-table tgw-rtb-123456789abcdef0
  %(prog)s 10.0.0.0/8 --network-profile my-network-profile --org-profile my-org-profile

Required AWS Permissions:
  - ec2:SearchTransitGatewayRoutes
  - ec2:DescribeTransitGatewayAttachments
  - organizations:DescribeAccount (optional, for account name lookup)
        """
    )

    parser.add_argument(
        'cidr_block',
        help='CIDR block to search for (e.g., 10.0.0.0/16)',
        nargs='?'
    )

    parser.add_argument(
        '--route-table', '-r',
        default='tgw-rtb-057244a1527cfcf2c',
        help='Transit Gateway Route Table ID (default: tgw-rtb-057244a1527cfcf2c)'
    )

    parser.add_argument(
        '--network-profile', '-n',
        default='network',
        help='AWS profile for EC2 operations (default: network)'
    )

    parser.add_argument(
        '--org-profile', '-o',
        default='root',
        help='AWS profile for Organizations operations (default: root)'
    )

    parser.add_argument(
        '--version', '-v',
        action='version',
        version='%(prog)s 1.1'
    )

    args = parser.parse_args()

    # Get CIDR block from argument or prompt user
    if args.cidr_block:
        cidr_to_check = args.cidr_block
    else:
        try:
            cidr_to_check = input("Enter the CIDR block to check (e.g., 10.0.0.0/16): ").strip()
            if not cidr_to_check:
                print("Error: No CIDR block provided.")
                sys.exit(1)
        except KeyboardInterrupt:
            print("\nOperation cancelled by user.")
            sys.exit(0)

    if not args.route_table:
        print("Error: Transit Gateway Route Table ID is required.")
        sys.exit(1)

    print(f"Checking CIDR block: {cidr_to_check}")
    print(f"Transit Gateway Route Table: {args.route_table}")
    print("-" * 50)

    result = check_cidr_and_get_account(
        args.route_table,
        cidr_to_check,
        args.network_profile,
        args.org_profile
    )

    if result is None:
        sys.exit(1)
    else:
        print("-" * 50)
        print(f"✓ Successfully found {len(result)} route(s) for CIDR {cidr_to_check}")


if __name__ == "__main__":
    main()