Source code for ecs_composex.vpc.vpc_template

# SPDX-License-Identifier: MPL-2.0
# Copyright 2020-2022 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,
    NoValue,
    Ref,
    Sub,
    Tags,
)
from troposphere.ec2 import VPC as VPCType
from troposphere.ec2 import (
    DHCPOptions,
    FlowLog,
    InternetGateway,
    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
from ecs_composex.iam import service_role_trust_policy
from ecs_composex.vpc import metadata
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_vpc_core(template, vpc_cidr, dhcp_options: dict): """ 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, ) 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=( dhcp_options["DomainName"] if (dhcp_options and "DomainName" in dhcp_options) else NoValue ), 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", ) )