Python AWS CDK中检测指定安全组是否存在并动态配置入站规则

需求概述 #

• 检测目标账户中指定名称的安全组是否存在
• 若该安全组存在，为指定安全组添加允许来自该组流量的入站规则
• 若不存在，跳过该规则，继续部署剩余栈

尝试的方法及问题 #

尝试通过from_lookup_by_name结合try-except捕获异常实现需求，但CDK未抛出可捕获的Python异常，执行cdk deploy或cdk synth时直接终止流程并报错：

尝试代码 #

try:
  sg = ec2.SecurityGroup.from_lookup_by_name(
            self,
            "SG",
            vpc=vpc,
            security_group_name="mysg",
        )
except:
  sg = None

报错信息 #

[Error at /stack/TestIngress] No security groups found matching {"account":"...","region":"eu-central-1","securityGroupName":"mysg","vpcId":"...","lookupRoleArn":"..."}

问题根源 #

CDK的from_lookup_by_name属于合成阶段（synth）查找操作，会在代码合成时提前调用AWS API验证资源存在性，该错误是CDK框架层面的验证错误，不属于Python运行时异常，因此无法通过try-except捕获。

解决方案 #

方法1：用自定义资源动态检测安全组（推荐） #

通过AwsCustomResource调用EC2 API在部署阶段动态查询安全组状态，根据结果条件化创建入站规则，不会阻断合成流程：

from aws_cdk import (
    Stack,
    ec2,
    custom_resources as cr,
    aws_iam as iam,
)
from constructs import Construct

class MyStack(Stack):
    def __init__(self, scope: Construct, construct_id: str, **kwargs) -> None:
        super().__init__(scope, construct_id, **kwargs)
        
        # 假设已通过lookup或参数获取目标VPC
        vpc = ec2.Vpc.from_lookup(self, "TargetVpc", vpc_id="vpc-xxxxxx")
        
        # 自定义资源：查询指定名称和VPC下的安全组
        sg_lookup = cr.AwsCustomResource(
            self, "SgLookup",
            on_create=cr.AwsSdkCall(
                service="EC2",
                action="describeSecurityGroups",
                parameters={
                    "Filters": [
                        {"Name": "group-name", "Values": ["mysg"]},
                        {"Name": "vpc-id", "Values": [vpc.vpc_id]}
                    ]
                },
                physical_resource_id=cr.PhysicalResourceId.of("SgLookupId")
            ),
            policy=cr.AwsCustomResourcePolicy.from_statements([
                iam.PolicyStatement(
                    actions=["ec2:DescribeSecurityGroups"],
                    resources=["*"]
                )
            ])
        )
        
        # 提取查询结果中的安全组ID（不存在则返回空）
        sg_id = sg_lookup.get_response_field("SecurityGroups.0.GroupId")
        
        # 目标安全组（需要添加入站规则的组）
        my_target_sg = ec2.SecurityGroup(
            self, "MyTargetSg",
            vpc=vpc,
            allow_all_outbound=True,
            description="Target SG to add conditional ingress rules"
        )
        
        # 条件化添加入站规则：仅当安全组存在时创建
        if sg_id:
            my_target_sg.add_ingress_rule(
                peer=ec2.Peer.security_group_id(sg_id),
                connection=ec2.Port.all_traffic(),  # 根据实际需求修改端口范围
                description="Allow traffic from existing SG 'mysg'"
            )

方法2：通过上下文参数提前传递状态 #

适合提前知道安全组状态的场景，先通过外部工具（如AWS CLI）检测安全组是否存在，再通过CDK上下文参数传递结果：

1. 部署时传入参数：

# 先通过CLI检测安全组是否存在，再传递参数
cdk deploy --context sg_exists=true --context sg_id=sg-xxxxxx

2. 代码中读取上下文并判断：

from aws_cdk import Stack, ec2
from constructs import Construct

class MyStack(Stack):
    def __init__(self, scope: Construct, construct_id: str, **kwargs) -> None:
        super().__init__(scope, construct_id, **kwargs)
        
        vpc = ec2.Vpc.from_lookup(self, "TargetVpc", vpc_id="vpc-xxxxxx")
        my_target_sg = ec2.SecurityGroup(
            self, "MyTargetSg",
            vpc=vpc,
            allow_all_outbound=True
        )
        
        # 读取上下文参数
        sg_exists = self.node.try_get_context("sg_exists") == "true"
        sg_id = self.node.try_get_context("sg_id")
        
        if sg_exists and sg_id:
            my_target_sg.add_ingress_rule(
                peer=ec2.Peer.security_group_id(sg_id),
                connection=ec2.Port.tcp(22),  # 示例端口
                description="Allow SSH from specified SG"
            )

内容的提问来源于stack exchange，提问作者Nathan