Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_check_zip_file(self):
positive_tests = [
'a'*4096,
Join('', ['a'*4096]),
Join('', ['a', 10]),
Join('', ['a'*4096, Ref('EmptyParameter')]),
Join('ab', ['a'*2047, 'a'*2047]),
GetAtt('foo', 'bar'),
]
for z in positive_tests:
Code.check_zip_file(z)
negative_tests = [
'a'*4097,
Join('', ['a'*4097]),
Join('', ['a'*4097, Ref('EmptyParameter')]),
Join('abc', ['a'*2047, 'a'*2047]),
]
for z in negative_tests:
with self.assertRaises(ValueError):
Code.check_zip_file(z)
def mkoutput(title, desc, val):
if isinstance(val, tuple):
val = GetAtt(val[0], val[1])
return Output(title, Description=desc, Value=val)
def render_ext_volume(context, context_ext, template, actual_ec2_instances, node=1):
vtype = context_ext.get('type', 'standard')
if node in actual_ec2_instances:
availability_zone = GetAtt(EC2_TITLE_NODE % node, "AvailabilityZone")
else:
availability_zone = context['aws']['availability-zone'] if node % 2 == 1 else context['aws']['redundant-availability-zone']
args = {
"Size": str(context_ext['size']),
# TODO: change
"AvailabilityZone": availability_zone,
"VolumeType": vtype,
"Tags": instance_tags(context, node),
}
ec2v = ec2.Volume(EXT_TITLE % node, **args)
template.add_resource(ec2v)
if node in actual_ec2_instances:
args = {
"InstanceId": Ref(EC2_TITLE_NODE % node),
Value=GetAtt(ec2_instance, "AvailabilityZone"),
),
Output(
"PublicIP",
Description="Public IP address of the newly created EC2 instance",
Value=GetAtt(ec2_instance, "PublicIp"),
),
Output(
"PrivateIP",
Description="Private IP address of the newly created EC2 instance",
Value=GetAtt(ec2_instance, "PrivateIp"),
),
Output(
"PublicDNS",
Description="Public DNSName of the newly created EC2 instance",
Value=GetAtt(ec2_instance, "PublicDnsName"),
),
Output(
"PrivateDNS",
Description="Private DNSName of the newly created EC2 instance",
Value=GetAtt(ec2_instance, "PrivateDnsName"),
),
])
print(template.to_json())
Tags = Tags(
Name = longname,
),
))
## CF Template Outputs
template.add_output([
Output(
"MongoDBInstanceID",
Description = "MongoDBInstance ID",
Value = Ref(mongodb_instance)
),
Output(
"MongoDBInstanceDNS",
Description = "MongoDBInstance Public DNS",
Value = GetAtt(mongodb_instance, "PublicDnsName")
)
])
print("## CloudFormation Template Built")
return template.to_json()
record_set_index += 1
if self.enabled == True:
self.create_output(
title='LoadBalancerArn',
value=troposphere.Ref(alb_resource),
ref=alb_config.paco_ref_parts + '.arn'
)
self.create_output(
title='LoadBalancerName',
value=troposphere.GetAtt(alb_resource, 'LoadBalancerName'),
ref=alb_config.paco_ref_parts + '.name'
)
self.create_output(
title='LoadBalancerFullName',
value=troposphere.GetAtt(alb_resource, 'LoadBalancerFullName'),
ref=alb_config.paco_ref_parts + '.fullname'
)
self.create_output(
title='LoadBalancerCanonicalHostedZoneID',
value=troposphere.GetAtt(alb_resource, 'CanonicalHostedZoneID'),
ref=alb_config.paco_ref_parts + '.canonicalhostedzoneid'
)
self.create_output(
title='LoadBalancerDNSName',
value=troposphere.GetAtt(alb_resource, 'DNSName'),
ref=alb_config.paco_ref_parts + '.dnsname',
)
if self.paco_ctx.legacy_flag('route53_record_set_2019_10_16') == False:
route53_ctl = self.paco_ctx.get_controller('route53')
for alb_dns in alb_config.dns:
def get_db_endpoint(self):
endpoint = GetAtt(DBINSTANCE, "Endpoint.Address")
return endpoint
))
###
#
# Outputs
#
###
template.add_output(Output(
"vpcId",
Value=Ref(cerberus_vpc)
))
template.add_output(Output(
"gatewayIamRoleArn",
Value=GetAtt(gateway_iam_role, "Arn")
))
template.add_output(Output(
"cmsIamRoleArn",
Value=GetAtt(cms_iam_role, "Arn")
))
template.add_output(Output(
"consulIamRoleArn",
Value=GetAtt(consul_iam_role, "Arn")
))
template.add_output(Output(
"vaultIamRoleArn",
Value=GetAtt(vault_iam_role, "Arn")
))
"",
[
"arn:aws:s3:::",
Ref(config_bucket),
"/config/secrets.json"
]
]
}
]
},
{
"Sid": "Allow-Bucket-Access-For-Consul",
"Effect": "Allow",
"Principal": {
"AWS": [
GetAtt(consul_iam_role, "Arn")
]
},
"Action": [
"s3:*"
],
"Resource": [
{
"Fn::Join": [
"",
[
"arn:aws:s3:::",
Ref(config_bucket),
"/data/consul/*"
]
]
},
app_name=self.app_name,
stage_name=self.stage_name),
PolicyDocument=Policy(
Version="2012-10-17",
Statement=self.get_statement_list(),
),
)
)
self.t.add_output(Output(
"AccessKey",
Value=Ref(zappa_user_keys),
Description="AWSAccessKeyId of new user",
))
self.t.add_output(Output(
"SecretKey",
Value=GetAtt(zappa_user_keys, "SecretAccessKey"),
Description="AWSSecretKey of new user",
))
return self.t.to_json()