def __init_

def init()

in centralized_vpc_endpoints/centralized_vpc_endpoints_stack.py [0:0]
75 lines of code
2 McCabe index (conditional complexity)

    def __init__(self, scope: cdk.Construct, construct_id: str, services, **kwargs) -> None:
        super().__init__(scope, construct_id, **kwargs)

        vpc_id = core.CfnParameter(
            self,
            "VPCId",
            description="The VPC where you want to place the endpoints",
            type="AWS::EC2::VPC::Id",
            allowed_pattern="^vpc-.*$",
        ).value_as_string

        org_cidr = core.CfnParameter(
            self,
            "OrgCIDR",
            description="The CIDR range that requests can originate from to use this endpoint.",
            allowed_pattern=r"^[\d]{1,3}\.[\d]{1,3}\.[\d]{1,3}\.[\d]{1,3}\/[\d]{1,3}$",
            constraint_description="CIDR should be in form X.X.X.X/X",
        ).value_as_string

        org_id = core.CfnParameter(
            self,
            "OrgID",
            description="The Organization ID which you'll allow to associate with your Hosted Zone",
            allowed_pattern="^o-[a-z0-9]{10,32}",
        ).value_as_string

        subnet_ids = core.CfnParameter(
            self,
            "EndpointSubnetIdList",
            type="List<AWS::EC2::Subnet::Id>",
            description="The subnets to insert VPC endpoints into",
            min_length=1,
            allowed_pattern="^subnet-.*$",
        ).value_as_list

        vpc = ec2.Vpc.from_vpc_attributes(
            self, "EndpointVPC", vpc_id=vpc_id, availability_zones=core.Fn.get_azs(core.Aws.REGION)
        )

        security_group_for_endpoints = ec2.SecurityGroup(self, "SecurityGroupForEndpoints", vpc=vpc)

        security_group_for_endpoints.add_ingress_rule(
            ec2.Peer.ipv4(org_cidr),
            connection=ec2.Port(protocol=ec2.Protocol.TCP, from_port=443, to_port=443, string_representation="TTA"),
        )

        # Create a Role that is assumed by the Spoke
        r53_role = iam.Role(
            self,
            "R53 Role",
            assumed_by=iam.AnyPrincipal().with_conditions({"StringEquals": {"aws:PrincipalOrgID": org_id}}),
        )

        # Add permissions to the Role
        r53_role.add_to_policy(
            iam.PolicyStatement(
                effect=iam.Effect.ALLOW,
                resources=["*"],
                actions=["route53:CreateVPCAssociationAuthorization", "route53:DeleteVPCAssociationAuthorization"],
            )
        )

        # The list of services are in the app.py file
        for service in services:
            record_name = f"{service}.{core.Aws.REGION}.amazonaws.com"

            endpoint = ec2.CfnVPCEndpoint(
                self,
                f"EndpointFor{service}",
                vpc_id=vpc_id,
                security_group_ids=[security_group_for_endpoints.security_group_id],
                vpc_endpoint_type="Interface",
                private_dns_enabled=False,
                service_name=f"com.amazonaws.{core.Aws.REGION}.{service}",
                subnet_ids=subnet_ids,
            )

            hosted_zone = route53.PrivateHostedZone(self, f"PrivateZoneFor{service}", zone_name=record_name, vpc=vpc)
            route53.RecordSet(
                self,
                f"AliasRecordFor{service}",
                record_type=route53.RecordType.A,
                zone=hosted_zone,
                record_name=record_name,
                target=route53.RecordTarget(
                    alias_target=RemoteInterfaceEndpointTarget(core.Fn.select(0, endpoint.attr_dns_entries))
                ),
            )

            core.CfnOutput(self, f"Route53DomainIDFor{service.upper()}", value=hosted_zone.hosted_zone_id)
        core.CfnOutput(self, "R53HubRoleToAssume", value=r53_role.role_arn)
def __init__()

def init()
