Используйте Boto3, чтобы найти маршрут подсетей через виртуальный частный шлюз

Question

Необходимо найти способ идентификации подсетей AWS VPC, которые маршрутизируют через виртуальный частный шлюз, с помощью Python Boto3. Другими словами, как я могу использовать python boto3 для идентификации частных подсетей в VPC?

Цель состоит в том, чтобы создать функцию Lambda, которая будет идентифицировать частные подсети в данном VPC, а затем запустить другую функцию Lambda вэти частные подсети.

Ниже приведен код, который я получил до сих пор. В нем перечислены все подсети в пределах VPC, к которому подключен виртуальный частный шлюз.

import boto3

def get_vpn_gateways():
    ec2_client = boto3.client('ec2')
    response = ec2_client.describe_vpn_gateways()
    return response

def get_vpc_subnets(VpcId):
    ec2 = boto3.resource('ec2')
    vpc = ec2.Vpc(VpcId)
    subnets = vpc.subnets.all()
    return subnets

# Get VPC Ids associated with the virtual private gateway
vpc_list = []
virtual_gateways = get_vpn_gateways() 
for virtual_gateway in virtual_gateways["VpnGateways"]:
    vgwId = virtual_gateway["VpnGatewayId"]
    vpcAttach = virtual_gateway["VpcAttachments"]
    vpc_list.append(vpcAttach[0]["VpcId"])
for vpc in vpc_list:
    print(vpc)
    subnets = get_vpc_subnets(vpc)
    for subnet in subnets:
        print(subnet)

Код до сих пор перечисляет все подсети в пределах VPC. Я думаю использовать маршрутизацию в качестве ключевого идентификатора для частной подсети. Если есть маршруты, проходящие через VGW, то я буду считать подсеть частной. Это имеет смысл?

Answer 1

Я думаю, что маршрутизация для 0.0.0.0/0 не является интернет-шлюзом, тогда это частная подсеть. Частная подсеть может быть направлена ​​к шлюзу NAT или виртуальному шлюзу, но не напрямую к интернет-шлюзу. Итак, я написал код, как показано ниже.

import boto3

ec2 = boto3.resource('ec2')
route_tables = ec2.route_tables.all()

for route_table in route_tables:
    for ra in route_table.routes_attribute:
        if ra.get('DestinationCidrBlock') == '0.0.0.0/0' and ra.get('GatewayId') is None:
            for rs in route_table.associations_attribute:
                if rs.get('SubnetId') is not None:
                    print(rs.get('SubnetId'))

Answer 2

Вот окончательный рабочий код, который ищет частные подсети в каждом VPC, к которому подключен Virtual Private Gateway. Он проверяет, находятся ли частные подсети в списке подсетей VPC, а затем продолжает их сохранение для другой функции Lambda. Это, наверное, не самый эффективный / действенный способ достижения моей цели. Стремится увидеть любые другие лучшие решения.

import boto3

def get_vpn_gateways():
    ec2_client = boto3.client('ec2')
    response = ec2_client.describe_vpn_gateways()
    return response

def get_vpc_subnets(VpcId):
    ec2 = boto3.resource('ec2')
    vpc = ec2.Vpc(VpcId)
    subnets = vpc.subnets.all()
    return subnets

def get_private_subnets():
    priv_subnet_list = []
    ec2 = boto3.resource('ec2')
    route_tables = ec2.route_tables.all()
    for route_table in route_tables:
        for ra in route_table.routes_attribute:
            if ra.get('DestinationCidrBlock') == '0.0.0.0/0' and ra.get('GatewayId') is None:
                for rs in route_table.associations_attribute:
                    if rs.get('SubnetId') is not None:
                        priv_subnet_list.append(rs.get('SubnetId'))
    return priv_subnet_list
def lambda_handler(event, context):
    vpc_list = []
    vpc_subnet_list = []
    virtual_gateways = get_vpn_gateways()
    lambda_subnets = []
    # Get VPC Ids associated with the virtual private gateway
    for virtual_gateway in virtual_gateways["VpnGateways"]:
        vgwId = virtual_gateway["VpnGatewayId"]
        vpcAttach = virtual_gateway["VpcAttachments"]
        vpc_list.append(vpcAttach[0]["VpcId"])
    # Get subnets within the VPC
    for vpc in vpc_list:
        subnets = get_vpc_subnets(vpc)
        for subnet in subnets:
            vpc_subnet_list.append(subnet.id)
        # Get Private subnets from the subnet list
        for privsubnet in get_private_subnets():
            if privsubnet in vpc_subnet_list:
                lambda_subnets.append(privsubnet)