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_allow_ref_task_role_arn(self):
task_definition = ecs.TaskDefinition(
"mytaskdef",
ContainerDefinitions=[
ecs.ContainerDefinition(
Image="myimage",
Memory="300",
Name="mycontainer",
)
],
TaskRoleArn=Ref(iam.Role("myRole"))
)
task_definition.to_dict()
def create_role(self):
t = self.template
statements = [
Statement(
Principal=Principal('Service', ['firehose.amazonaws.com']),
Effect=Allow,
Action=[sts.AssumeRole],
Condition=Condition(
StringEquals('sts:ExternalId', Ref('AWS::AccountId')),
),
),
]
firehose_role_policy = Policy(Statement=statements)
t.add_resource(
iam.Role(
IAM_ROLE,
AssumeRolePolicyDocument=firehose_role_policy,
Path='/',
Policies=self.generate_iam_policies(),
),
)
t.add_output(Output('Role', Value=Ref(IAM_ROLE)))
t.add_output(Output('RoleArn', Value=GetAtt(IAM_ROLE, 'Arn')))
def setup_self_terminating_stack(self):
# From https://aws.amazon.com/blogs/devops/scheduling-automatic-deletion-of-application-environments/
self._resources.update({
'StackDeletorRole': iam.Role(
'StackDeletorRole',
Metadata={
'Description': 'Some comment',
},
AssumeRolePolicyDocument={
"Statement": [{
"Effect": "Allow",
"Principal": {
"Service": 'ec2.amazonaws.com',
},
"Action": ["sts:AssumeRole"]
}]
},
Path='/',
Policies=[
iam.Policy(
def create_policy(self, name):
statements = self.generate_policy_statements()
if not statements:
return
t = self.template
policy = t.add_resource(
iam.PolicyType(
"{}Policy".format(name),
PolicyName=Sub("${AWS::StackName}-${Name}-policy", Name=name),
PolicyDocument=Policy(
Statement=statements,
),
Roles=[Ref(role) for role in self.roles],
)
)
t.add_output(
Output(name + "PolicyName", Value=Ref(policy))
)
self.policies.append(policy)
# EventRule that is invoked when ECR image is tagged
events_rule_role_resource = troposphere.iam.Role(
title='EventsRuleRole',
template=self.template,
AssumeRolePolicyDocument=Policy(
Version='2012-10-17',
Statement=[
Statement(
Effect=Allow,
Action=[AssumeRole],
Principal=Principal('Service',['events.amazonaws.com'])
)
],
),
Policies=[
troposphere.iam.Policy(
PolicyName="TargetInvocation",
PolicyDocument=Policy(
Version='2012-10-17',
Statement=[
Statement(
Effect=Allow,
Action=[awacs.codepipeline.StartPipelineExecution],
Resource=[self.pipeline_arn],
)
]
)
)
],
)
event_rule_name = create_event_rule_name(self.resource)
es_policies = [iam.Policy(
PolicyName='sqsWrite',
PolicyDocument={
"Statement": [{
"Effect" : "Allow",
"Action" : ["sqs:ChangeMessageVisibility","sqs:ChangeMessageVisibilityBatch","sqs:GetQueueAttributes","sqs:GetQueueUrl","sqs:ListQueues","sqs:SendMessage","sqs:SendMessageBatch"],
"Resource" : [GetAtt(logging_queue, 'Arn')]}]}),
iam.Policy(
PolicyName='ec2DescribeAllInstancesInRegion',
PolicyDocument={
"Statement": [{
"Effect" : "Allow",
"Action" :["ec2:Describe*"],
"Resource" : "*"}]}),
iam.Policy(
PolicyName='s3AllForBackupBucket',
PolicyDocument={
"Statement": [{
"Effect" : "Allow",
"Action" : ["s3:*"],
"Resource" : [Join('', ['arn:aws:s3:::', Ref(backup_bucket), "/*"])]}]}),
iam.Policy(
PolicyName='s3ListAndGetBucket',
PolicyDocument={
"Statement" : [{
"Effect" : "Allow",
"Action" : ["s3:List*", "s3:GetBucket*"],
"Resource" : "arn:aws:s3:::*"}]})]
iam_profile = self.create_instance_profile('elasticsearch', es_policies)
def add_lambda_role(template: troposphere.Template) -> troposphere.iam.Role:
role = troposphere.iam.Role(
f"LambdaRole", template,
AssumeRolePolicyDocument={
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Principal": {
"Service": [
"lambda.amazonaws.com"
]
},
"Action": [
"sts:AssumeRole"
]
}
],
sshlocation = template.add_parameter(Parameter(
'SSHLocation',
Description='The IP address range that can be used to SSH to'
' the EC2 instances',
Type='String',
MinLength='9',
MaxLength='18',
Default='0.0.0.0/0',
AllowedPattern='(\\d{1,3})\\.(\\d{1,3})\\.'
'(\\d{1,3})\\.(\\d{1,3})/(\\d{1,2})',
ConstraintDescription='must be a valid IP CIDR range of the'
' form x.x.x.x/x.'
))
# Resources
webserverrole = template.add_resource(iam.Role(
'WebServerRole',
AssumeRolePolicyDocument=PolicyDocument(
Statement=[
Statement(
Effect=Allow,
Action=[AssumeRole],
Principal=Principal('Service',
[FindInMap('Region2Principal',
Ref('AWS::Region'),
'EC2Principal')]),
)
]
),
Path='/',
))
}
]
}
"""
slm_iam_role = t.add_resource(iam.Role(
"SLMServerRole",
AssumeRolePolicyDocument=Policy(
Statement=[
Statement(
Effect=Allow,
Action=[AssumeRole],
Principal=Principal("Service", ["ec2.amazonaws.com"])
)
]
),
Policies=[iam.Policy(
PolicyName="SLMServerPolicy",
PolicyDocument= Policy(
Statement=[
Statement(
Effect=Allow,
Action=[
Action('s3', 'List*'),
Action('s3', 'Get*'),
],
Resource=['arn:aws:s3:::*']
)]
)
)
]
))
],
KeySchema=[
dynamodb.KeySchema(
AttributeName="group",
KeyType="HASH",
)
],
))
template.add_output(Output(
"GroupTableName",
Description="DynamoDB table for groups",
Value=Ref(group_table),
Export=Export(Join('-', [Ref(AWS_STACK_NAME), 'GroupTable'])),
))
lambda_role = template.add_resource(iam.Role(
"LambdaRole",
Path="/",
AssumeRolePolicyDocument={
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
# Lambda@Edge uses a different principal than normal lambda
"Principal": {
"Service": [
"lambda.amazonaws.com",
"edgelambda.amazonaws.com"
]
},
"Action": "sts:AssumeRole"
}