此回复是 AWS 支持对我与 OP 问题类似的询问的编辑回复。
此功能当前不可用,AWS 的 CloudFormation 开发团队已意识到此问题,并且已提出功能请求。
作为一种解决方法,您可以利用 Lambda 支持的自定义资源来获取安全 ID 并将其传递给自定义资源,以便可以在 CF 堆栈中访问它。
在这种方法中,您将创建一个 Lambda 函数,该函数可以将安全组名称和 VPC-id 作为输入,并将安全组 ID 作为输出。创建的自定义资源是一段代码,它将使用组名称和 VPC-id 向 Lambda 函数发出信号。Lambda 将 security group-id 返回到这个自定义资源,您可以获取 sg-id 如下所示:
{ "Fn::GetAtt" : ["CustomResouce", "security_group_id"] }
此消息末尾包含一个示例模板和一个示例 Lambda 函数(用于获取安全组 ID)。在模板中,该函数已在代码的输出中使用,但您可以在客户端的 security_group 中使用相同的功能,如下所示:
"SourceSecurityGroupId" : { "Fn::GetAtt" : ["CustomResouce", "security_group_id"] },
"SourceSecurityGroupName" : { "Fn::Sub": [ "${Alias}_controllers", { "Alias": {"Ref" : "Alias" }} ]},
自定义 Lambda 函数customresouce.py(放置在 Lambda 可以访问的 S3 存储桶中):
import json
import boto3
import time
from botocore.vendored import requests
def lambda_handler(event, context):
print event['RequestType']
try:
if event['RequestType'] == 'Delete':
print "delete"
responseData = {'response': 'Delete'}
responseStatus = 'SUCCESS'
elif event['RequestType'] == 'Create':
print "blabla"
ec2 = boto3.resource('ec2')
vpc = ec2.Vpc(event['ResourceProperties']['vpc'])
security_group_iterator = vpc.security_groups.filter(GroupNames = [event['ResourceProperties']['security_group']])
sg_id = list(security_group_iterator.filter(GroupNames = [event['ResourceProperties']['security_group']]))[0].id
print sg_id
responseData = {'security_group_id': sg_id}
elif event['RequestType'] == 'Update':
print "update"
responseData = {'response': 'Update'}
responseStatus = 'SUCCESS'
responseStatus = 'SUCCESS'
except:
responseStatus = 'FAILED'
responseData = {'FAILED': 'Something bad happened.'}
sendResponse(event, context, responseStatus, responseData)
def sendResponse(event, context, responseStatus, responseData, reason=None, physical_resource_id=None):
responseBody = {'Status': responseStatus,
'Reason': 'See the details in CloudWatch Log Stream: ' + context.log_stream_name,
'PhysicalResourceId': physical_resource_id or context.log_stream_name,
'StackId': event['StackId'],
'RequestId': event['RequestId'],
'LogicalResourceId': event['LogicalResourceId'],
'Data': responseData}
print 'RESPONSE BODY:n' + json.dumps(responseBody)
responseUrl = event['ResponseURL']
json_responseBody = json.dumps(responseBody)
headers = {
'content-type' : '',
'content-length' : str(len(json_responseBody))
}
try:
response = requests.put(responseUrl,
data=json_responseBody,
headers=headers)
print "Status code: " + response.reason
except Exception as e:
print "send(..) failed executing requests.put(..): " + str(e)
使用自定义 Lambda 函数的示例模板:
{
"Resources": {
"myDirectory" : {
"Type" : "AWS::DirectoryService::SimpleAD",
"Properties" : {
"Name" : "corp.example.com",
"Password" : "P@ssword",
"Size" : "Small",
"VpcSettings" : {
"SubnetIds" : [ "subnet_value-1", "subnet_value-2" ],
"VpcId" : "your_vpc-id"
}
}
},
"CustomResouce": {
"DependsOn": "myDirectory",
"Type": "Custom::GettingsecuritygroupId",
"Version" : "1.0",
"Properties" : {
"ServiceToken": {"Fn::GetAtt" : ["Mylambda","Arn"]},
"vpc" : "vpc-b0ee43c9",
"security_group" : { "Fn::Sub": [ "${Alias}_controllers", {"Alias":{"Fn::GetAtt" : ["myDirectory","Alias"]} }]}
}
},
"Mylambda":{
"Type" : "AWS::Lambda::Function",
"Properties" : {
"Code" : {
"S3Bucket": "Your_s3_bucket_name",
"S3Key": "customresource.py.zip"
},
"Handler" : "customresource.lambda_handler",
"Role" : "Role_whic_has_permissions_ec2:*",
"Runtime" :"python2.7",
"Timeout" : "60"
}
}
},
"Outputs":{
"SGID" : {
"Value" : { "Fn::GetAtt" : ["CustomResouce", "security_group_id"] }
}
}
}
相当笨拙,但它似乎可以作为临时解决方案,直到 AWS 开始将功能实施到他们的 API/CloudFormation/Hosted AD 中。注意:我还没有机会测试上述内容,但我将其发布在此处,供 OP 和其他任何可能正在寻找此问题解决方案的人使用。
参考:
自定义资源参考
AWS Lambda 支持的自定义资源
AWS::Lambda::函数