def check_service_vpc_endpoints()

in MWAA/verify_env/verify_env.py [0:0]

• 48 lines of code
• 5 McCabe index (conditional complexity)

def check_service_vpc_endpoints(ec2_client, subnets):
    '''
    should be used if the environment does not have internet access through NAT Gateway
    '''
    top_level_domain = "com.amazonaws."
    service_endpoints = [
        top_level_domain + REGION + '.airflow.api',
        top_level_domain + REGION + '.airflow.env',
        top_level_domain + REGION + '.airflow.ops',
        top_level_domain + REGION + '.sqs',
        top_level_domain + REGION + '.ecr.api',
        top_level_domain + REGION + '.ecr.dkr',
        top_level_domain + REGION + '.kms',
        top_level_domain + REGION + '.s3',
        top_level_domain + REGION + '.monitoring',
        top_level_domain + REGION + '.logs'
    ]
    vpc_endpoints = ec2_client.describe_vpc_endpoints(Filters=[
        {
            'Name': 'service-name',
            'Values': service_endpoints
        },
        {
            'Name': 'vpc-id',
            'Values': [
                subnets[0]['VpcId']
            ]
        }
    ])['VpcEndpoints']
    # filter by subnet ids here, if the vpc endpoints include the env's subnet ids then check those
    s_ids = [subnet['SubnetId'] for subnet in subnets]
    vpc_endpoints = [endpoint for endpoint in vpc_endpoints if all(subnet in s_ids for subnet in
                     endpoint['SubnetIds'])]
    if len(vpc_endpoints) != 9:
        print("The route for the subnets do not have a NAT gateway." +
              "This suggests vpc endpoints are needed to connect to:")
        print('s3, ecr, kms, sqs, monitoring, airflow.api, airflow.env, airflow.ops')
        print("The environment's subnets currently have these endpoints: ")
        for endpoint in vpc_endpoints:
            print(endpoint['ServiceName'])
        print("The environment's subnets do not have these endpoints: ")
        vpc_service_endpoints = [e['ServiceName'] for e in vpc_endpoints]
        for i, service_endpoint in enumerate(service_endpoints):
            if service_endpoint not in vpc_service_endpoints:
                print(service_endpoint)
        check_vpc_endpoint_private_dns_enabled(vpc_endpoints)
    else:
        print("The route for the subnets do not have a NAT Gateway. However, there are sufficient VPC endpoints")