def load_ec2_vpcs()

in cartography/intel/aws/ec2.py [0:0]

• 25 lines of code
• 3 McCabe index (conditional complexity)

def load_ec2_vpcs(session, data, region, current_aws_account_id, aws_update_tag):
    # https://docs.aws.amazon.com/cli/latest/reference/ec2/describe-vpcs.html
    # {
    #     "Vpcs": [
    #         {
    #             "VpcId": "vpc-a01106c2",
    #             "InstanceTenancy": "default",
    #             "Tags": [
    #                 {
    #                     "Value": "MyVPC",
    #                     "Key": "Name"
    #                 }
    #             ],
    #             "CidrBlockAssociations": [
    #                 {
    #                     "AssociationId": "vpc-cidr-assoc-a26a41ca",
    #                     "CidrBlock": "10.0.0.0/16",
    #                     "CidrBlockState": {
    #                         "State": "associated"
    #                     }
    #                 }
    #             ],
    #             "State": "available",
    #             "DhcpOptionsId": "dopt-7a8b9c2d",
    #             "CidrBlock": "10.0.0.0/16",
    #             "IsDefault": false
    #         }
    #     ]
    # }

    ingest_vpc = """
    MERGE (new_vpc:AWSVpc{id: {VpcId}})
    ON CREATE SET new_vpc.firstseen = timestamp(), new_vpc.vpcid ={VpcId}
    SET new_vpc.instance_tenancy = {InstanceTenancy},
    new_vpc.state = {State},
    new_vpc.is_default = {IsDefault},
    new_vpc.primary_cidr_block = {PrimaryCIDRBlock},
    new_vpc.dhcp_options_id = {DhcpOptionsId},
    new_vpc.region = {Region},
    new_vpc.lastupdated = {aws_update_tag}
    WITH new_vpc
    MATCH (awsAccount:AWSAccount{id: {AWS_ACCOUNT_ID}})
    MERGE (awsAccount)-[r:RESOURCE]->(new_vpc)
    ON CREATE SET r.firstseen = timestamp()
    SET r.lastupdated = {aws_update_tag}"""

    for vpc in data['Vpcs']:
        vpc_id = vpc["VpcId"]  # fail if not present

        session.run(
            ingest_vpc,
            VpcId=vpc_id,
            InstanceTenancy=vpc.get("InstanceTenancy", None),
            State=vpc.get("State", None),
            IsDefault=vpc.get("IsDefault", None),
            PrimaryCIDRBlock=vpc.get("CidrBlock", None),
            DhcpOptionsId=vpc.get("DhcpOptionsId", None),
            Region=region,
            AWS_ACCOUNT_ID=current_aws_account_id,
            aws_update_tag=aws_update_tag)

        load_cidr_association_set(session,
                                  vpc_id=vpc_id,
                                  block_type="ipv4",
                                  vpc_data=vpc,
                                  aws_update_tag=aws_update_tag)

        load_cidr_association_set(session,
                                  vpc_id=vpc_id,
                                  block_type="ipv6",
                                  vpc_data=vpc,
                                  aws_update_tag=aws_update_tag)