broad-morning-80838
11/18/2022, 10:36 PMOutput[T]
. I'm trying to create a security group, then subsequently create security group rules using the id of the security group in the name of the rules. I've read through multiple threads here where people were trying to do similar things, and have tried to use solutions that worked for them (e.g. apply
etc) but am still stuck and very confused.
Here's the code in question:
# Create a security group for external ALBs
ext_alb_sg = SecurityGroup(
name="ext-alb-sg",
description="Security group for external load balancers",
)
ext_alb_sg_id = ext_alb_sg.id.apply(lambda id: id).apply(lambda id: f"{id}")
# Add rules to external ALB security group
ext_alb_inbound_rules = InboundSecurityGroupRules(
parent_sg_id=ext_alb_sg_id,
source_cidrs="0.0.0.0/0",
port=443,
)
ext_alb_outbound_rules = OutboundSecurityGroupRules(
parent_sg_id=ext_alb_sg_id,
destination_cidrs=VPC_CIDR,
port=80,
)
miniature-musician-31262
11/18/2022, 10:57 PMInboundSecurityGroupRules
?broad-morning-80838
11/18/2022, 11:02 PMfrom typing import List, Mapping, Optional, Sequence
import pulumi
import pulumi_aws as aws
from . import VPC_ID
class SecurityGroup(aws.ec2.SecurityGroup):
"""Creates a security group"""
def __init__(
self,
name: str,
description: str,
vpc_id: str = VPC_ID,
ingress: Optional[Sequence[aws.ec2.SecurityGroupIngressArgs]] = None,
egress: Optional[Sequence[aws.ec2.SecurityGroupEgressArgs]] = None,
tags: Optional[Mapping[str, str]] = None,
opts: Optional[pulumi.ResourceOptions] = None,
) -> None:
# Append "-sg" to name if not already present
if name.endswith("-sg"):
sg_name = name
else:
sg_name = f"{name}-sg"
super().__init__(
resource_name=sg_name,
name=sg_name,
description=description,
vpc_id=vpc_id,
ingress=ingress,
egress=egress,
opts=opts,
tags={"Name": sg_name},
)
pulumi.export(name, {"security_group_id": self.id})
class CIDRSecurityGroupRule(aws.ec2.SecurityGroupRule):
"""Creates a security group rule to/from a CIDR"""
def __init__(
self,
name: str,
security_group_id: str,
from_port: int,
to_port: int,
protocol: str,
type: str,
cidr_blocks: Optional[List[str]] = None,
opts: Optional[pulumi.ResourceOptions] = None,
) -> None:
sgr_name = f"{name}-sgr"
super().__init__(
resource_name=sgr_name,
security_group_id=security_group_id,
type=type,
from_port=from_port,
to_port=to_port,
protocol=protocol,
cidr_blocks=cidr_blocks,
)
pulumi.export(name, {"security_group_rule_id": self.id})
class SGSecurityGroupRule(aws.ec2.SecurityGroupRule):
"""Creates a security group rule to/from a security group"""
def __init__(
self,
name: str,
security_group_id: str,
from_port: int,
to_port: int,
protocol: str,
type: str,
source_security_group_id: Optional[str] = None,
opts: Optional[pulumi.ResourceOptions] = None,
) -> None:
sgr_name = f"{name}-sgr"
super().__init__(
resource_name=sgr_name,
security_group_id=security_group_id,
type=type,
from_port=from_port,
to_port=to_port,
protocol=protocol,
source_security_group_id=source_security_group_id,
)
pulumi.export(name, {"security_group_rule_id": self.id})
class InboundSecurityGroupRules:
"""Creates security group rules for one TCP port from one or more CIDRs and/or one security group"""
def __init__(
self,
parent_sg_id: str,
port: int,
source_cidrs: Optional[List[str]] = None,
source_sg_id: Optional[str] = None,
opts: Optional[pulumi.ResourceOptions] = None,
) -> None:
if source_cidrs is not None:
if isinstance(source_cidrs, str):
subnets = [source_cidrs]
rule_name_suffix = source_cidrs
else:
subnets = [subnet.value for subnet in source_cidrs]
rule_name_suffix = source_cidrs.__name__
self.cidr_rule = CIDRSecurityGroupRule(
name=f"{parent_sg_id}-{rule_name_suffix}",
cidr_blocks=subnets,
security_group_id=parent_sg_id,
from_port=port,
to_port=port,
opts=opts,
type="ingress",
protocol="tcp",
)
if source_sg_id is not None:
self.sg_rule = SGSecurityGroupRule(
name=f"{parent_sg_id}-{source_sg_id}",
source_security_group_id=source_sg_id,
security_group_id=parent_sg_id,
from_port=port,
to_port=port,
opts=opts,
type="ingress",
protocol="tcp",
)
if source_cidrs is None and source_sg_id is None:
raise ValueError("Must provide at least one of source_cidrs or source_sg_id")
class OutboundSecurityGroupRules:
"""Creates security group rules for one TCP port to one or more CIDRs and/or one security group"""
def __init__(
self,
parent_sg_id: str,
port: int,
destination_cidrs: Optional[List[str]] = None,
destination_sg_id: Optional[str] = None,
opts: Optional[pulumi.ResourceOptions] = None,
) -> None:
if destination_cidrs is not None:
if isinstance(destination_cidrs, str):
subnets = [destination_cidrs]
rule_name_suffix = destination_cidrs
else:
subnets = [subnet.value for subnet in destination_cidrs]
rule_name_suffix = destination_cidrs.__name__
self.cidr_rule = CIDRSecurityGroupRule(
name=f"{parent_sg_id}-{rule_name_suffix}",
cidr_blocks=subnets,
security_group_id=parent_sg_id,
from_port=port,
to_port=port,
type="egress",
protocol="tcp",
)
if destination_sg_id is not None:
self.sg_rule = SGSecurityGroupRule(
name=f"{parent_sg_id}-{destination_sg_id}",
source_security_group_id=destination_sg_id,
security_group_id=parent_sg_id,
from_port=port,
to_port=port,
type="egress",
protocol="tcp",
)
if destination_cidrs is None and destination_sg_id is None:
raise ValueError("Must provide at least one of destination_cidrs or destination_sg_id")
miniature-musician-31262
11/18/2022, 11:08 PMbroad-morning-80838
11/18/2022, 11:09 PMminiature-musician-31262
11/18/2022, 11:09 PMbroad-morning-80838
11/18/2022, 11:10 PMminiature-musician-31262
11/18/2022, 11:10 PMbroad-morning-80838
11/18/2022, 11:10 PMminiature-musician-31262
11/18/2022, 11:11 PMpulumi.ComponentResource
broad-morning-80838
11/18/2022, 11:20 PMminiature-musician-31262
11/18/2022, 11:20 PMbroad-morning-80838
11/18/2022, 11:23 PMInboundSecurityGroupRule
(for example) multiple times to handle multiple sets of CIDRs etc from various sources for one security group, so I may also need to refactor the class to handle that, or maybe that means I'll need to rethink the approach. In any case I think I understand some of this a bit better!