Source code for ecs_composex.vpc.vpc_template

#  -*- coding: utf-8 -*-
# SPDX-License-Identifier: MPL-2.0
# Copyright 2020-2021 John Mille <john@compose-x.io>

"""
Create the VPC template and its associated resources
"""

import re

from troposphere import (
    AWS_ACCOUNT_ID,
    AWS_NO_VALUE,
    AWS_PARTITION,
    AWS_REGION,
    GetAtt,
    If,
    Join,
    Ref,
    Sub,
    Tags,
)
from troposphere.ec2 import VPC as VPCType
from troposphere.ec2 import (
    DHCPOptions,
    Entry,
    FlowLog,
    InternetGateway,
    PrefixList,
    VPCDHCPOptionsAssociation,
    VPCGatewayAttachment,
)
from troposphere.iam import Policy, Role
from troposphere.logs import LogGroup

from ecs_composex.common.cfn_conditions import USE_STACK_NAME_CON_T
from ecs_composex.common.cfn_params import ROOT_STACK_NAME, ROOT_STACK_NAME_T
from ecs_composex.common.outputs import ComposeXOutput
from ecs_composex.dns.dns_params import PRIVATE_DNS_ZONE_NAME
from ecs_composex.iam import service_role_trust_policy
from ecs_composex.vpc import metadata, vpc_params
from ecs_composex.vpc.vpc_params import IGW_T, VPC_T

AZ_INDEX_PATTERN = r"(([a-z0-9-]+)([a-z]{1}$))"
AZ_INDEX_RE = re.compile(AZ_INDEX_PATTERN)


[docs]def add_template_outputs(template, vpc, storage_subnets, public_subnets, app_subnets): """ Function to add outputs / exports to the template :param template: VPC Template() :param vpc: Vpc() for Ref() :param storage_subnets: List of Subnet() :param public_subnets: List of Subnet() :param app_subnets: List of Subnet() """ template.add_output( ComposeXOutput( vpc, [ (vpc_params.VPC_ID_T, vpc_params.VPC_ID_T, Ref(vpc)), ( vpc_params.STORAGE_SUBNETS_T, vpc_params.STORAGE_SUBNETS_T, Join(",", [Ref(subnet) for subnet in storage_subnets]), ), ( vpc_params.PUBLIC_SUBNETS_T, vpc_params.PUBLIC_SUBNETS_T, Join(",", [Ref(subnet) for subnet in public_subnets]), ), ( vpc_params.APP_SUBNETS_T, vpc_params.APP_SUBNETS_T, Join(",", [Ref(subnet) for subnet in app_subnets]), ), ], duplicate_attr=True, ).outputs )
[docs]def add_vpc_cidrs_outputs(template, vpc, layers): """ Function to add outputs / exports to the template :param template: VPC Template() :type template: troposphere.Template :param layers: dict of layers CIDRs to export :type layers: dict """ template.add_output( ComposeXOutput( vpc, [ ( vpc_params.STORAGE_SUBNETS_CIDR_T, vpc_params.STORAGE_SUBNETS_CIDR_T, Join(",", [cidr for cidr in layers["stor"]]), ), ( vpc_params.PUBLIC_SUBNETS_CIDR_T, vpc_params.PUBLIC_SUBNETS_CIDR_T, Join(",", [cidr for cidr in layers["pub"]]), ), ( vpc_params.APP_SUBNETS_CIDR_T, vpc_params.APP_SUBNETS_CIDR_T, Join(",", [cidr for cidr in layers["app"]]), ), ], duplicate_attr=True, ).outputs )
[docs]def add_vpc_core(template, vpc_cidr): """ Function to create the core resources of the VPC and add them to the core VPC template :param template: VPC Template() :param vpc_cidr: str of the VPC CIDR i.e. 192.168.0.0/24 :return: tuple() with the vpc and igw object """ vpc = VPCType( VPC_T, template=template, CidrBlock=vpc_cidr, EnableDnsHostnames=True, EnableDnsSupport=True, Tags=Tags( Name=If( USE_STACK_NAME_CON_T, Ref("AWS::StackName"), Ref(ROOT_STACK_NAME), ), EnvironmentName=If( USE_STACK_NAME_CON_T, Ref("AWS::StackName"), Ref(ROOT_STACK_NAME), ), ), Metadata=metadata, ) template.add_resource( PrefixList( "VpcPrefixList", AddressFamily="IPv4", Entries=[ Entry( Cidr=vpc_cidr, Description=Sub(f"Primary CIDR for ${{{vpc.title}}}"), ) ], MaxEntries=5, PrefixListName=Ref(vpc), ) ) igw = InternetGateway(IGW_T, template=template) VPCGatewayAttachment( "VPCGatewayAttachement", template=template, InternetGatewayId=Ref(igw), VpcId=Ref(vpc), Metadata=metadata, ) dhcp_opts = DHCPOptions( "VpcDhcpOptions", template=template, DomainName=If( USE_STACK_NAME_CON_T, Sub( f"svc.${{AWS::StackName}}.${{{PRIVATE_DNS_ZONE_NAME.title}}} " f"${{AWS::StackName}}.${{{PRIVATE_DNS_ZONE_NAME.title}}}" ), Sub( f"svc.${{{ROOT_STACK_NAME_T}}}.${{{PRIVATE_DNS_ZONE_NAME.title}}} " f"${{{ROOT_STACK_NAME_T}}}.${{{PRIVATE_DNS_ZONE_NAME.title}}}" ), ), DomainNameServers=["AmazonProvidedDNS"], Tags=Tags(Name=Sub(f"dhcp-${{{vpc.title}}}")), Metadata=metadata, ) VPCDHCPOptionsAssociation( "VpcDhcpOptionsAssociate", template=template, DhcpOptionsId=Ref(dhcp_opts), VpcId=Ref(vpc), Metadata=metadata, ) return vpc, igw
[docs]def add_vpc_flow(template, vpc, boundary=None): """ Function to add VPC Flow Log to log VPC :param troposphere.Template template: :param vpc: The VPC Object :param str boundary: """ if boundary and boundary.startswith("arn:aws"): perm_boundary = boundary elif boundary and not boundary.startswith("arn:aws"): perm_boundary = Sub( f"arn:${{{AWS_PARTITION}}}:iam:${{{AWS_REGION}}}:${{{AWS_ACCOUNT_ID}}}:policy/{boundary}" ) else: perm_boundary = Ref(AWS_NO_VALUE) log_group = template.add_resource( LogGroup( "FlowLogsGroup", RetentionInDays=14, LogGroupName=Sub(f"flowlogs/vpc/${{{vpc.title}}}"), ) ) role = template.add_resource( Role( "FlowLogsRole", AssumeRolePolicyDocument=service_role_trust_policy("ec2"), PermissionsBoundary=perm_boundary, Policies=[ Policy( PolicyName="CloudWatchAccess", PolicyDocument={ "Version": "2012-10-17", "Statement": [ { "Sid": "AllowCloudWatchLoggingToSpecificLogGroup", "Effect": "Allow", "Action": [ "logs:CreateLogStream", "logs:PutLogEvents", ], "Resource": GetAtt(log_group, "Arn"), } ], }, ) ], ) ) template.add_resource( FlowLog( "VpcFlowLogs", DeliverLogsPermissionArn=GetAtt(role, "Arn"), LogGroupName=Ref(log_group), LogDestinationType="cloud-watch-logs", MaxAggregationInterval=600, ResourceId=Ref(vpc), ResourceType="VPC", TrafficType="ALL", ) )