1from samtranslator.model import PropertyType, Resource
2from samtranslator.model.types import is_type, is_str, list_of
3from samtranslator.model.intrinsics import ref, fnGetAtt
4
5
6class IAMRole(Resource):
7    resource_type = "AWS::IAM::Role"
8    property_types = {
9        "AssumeRolePolicyDocument": PropertyType(True, is_type(dict)),
10        "ManagedPolicyArns": PropertyType(False, is_type(list)),
11        "Path": PropertyType(False, is_str()),
12        "Policies": PropertyType(False, is_type(list)),
13        "PermissionsBoundary": PropertyType(False, is_str()),
14        "Tags": PropertyType(False, list_of(is_type(dict))),
15    }
16
17    runtime_attrs = {"name": lambda self: ref(self.logical_id), "arn": lambda self: fnGetAtt(self.logical_id, "Arn")}
18
19
20class IAMRolePolicies:
21    @classmethod
22    def construct_assume_role_policy_for_service_principal(cls, service_principal):
23        document = {
24            "Version": "2012-10-17",
25            "Statement": [
26                {
27                    "Action": ["sts:AssumeRole"],
28                    "Effect": "Allow",
29                    "Principal": {"Service": [service_principal]},
30                }
31            ],
32        }
33        return document
34
35    @classmethod
36    def step_functions_start_execution_role_policy(cls, state_machine_arn, logical_id):
37        document = {
38            "PolicyName": logical_id + "StartExecutionPolicy",
39            "PolicyDocument": {
40                "Statement": [{"Action": "states:StartExecution", "Effect": "Allow", "Resource": state_machine_arn}]
41            },
42        }
43        return document
44
45    @classmethod
46    def stepfunctions_assume_role_policy(cls):
47        document = {
48            "Version": "2012-10-17",
49            "Statement": [
50                {
51                    "Action": ["sts:AssumeRole"],
52                    "Effect": "Allow",
53                    "Principal": {"Service": ["states.amazonaws.com"]},
54                }
55            ],
56        }
57        return document
58
59    @classmethod
60    def cloud_watch_log_assume_role_policy(cls):
61        document = {
62            "Version": "2012-10-17",
63            "Statement": [
64                {
65                    "Action": ["sts:AssumeRole"],
66                    "Effect": "Allow",
67                    "Principal": {"Service": ["apigateway.amazonaws.com"]},
68                }
69            ],
70        }
71        return document
72
73    @classmethod
74    def lambda_assume_role_policy(cls):
75        document = {
76            "Version": "2012-10-17",
77            "Statement": [
78                {"Action": ["sts:AssumeRole"], "Effect": "Allow", "Principal": {"Service": ["lambda.amazonaws.com"]}}
79            ],
80        }
81        return document
82
83    @classmethod
84    def dead_letter_queue_policy(cls, action, resource):
85        """Return the DeadLetterQueue Policy to be added to the LambdaRole
86        :returns: Policy for the DeadLetterQueue
87        :rtype: Dict
88        """
89        return {
90            "PolicyName": "DeadLetterQueuePolicy",
91            "PolicyDocument": {
92                "Version": "2012-10-17",
93                "Statement": [{"Action": action, "Resource": resource, "Effect": "Allow"}],
94            },
95        }
96
97    @classmethod
98    def sqs_send_message_role_policy(cls, queue_arn, logical_id):
99        document = {
100            "PolicyName": logical_id + "SQSPolicy",
101            "PolicyDocument": {"Statement": [{"Action": "sqs:SendMessage", "Effect": "Allow", "Resource": queue_arn}]},
102        }
103        return document
104
105    @classmethod
106    def sns_publish_role_policy(cls, topic_arn, logical_id):
107        document = {
108            "PolicyName": logical_id + "SNSPolicy",
109            "PolicyDocument": {"Statement": [{"Action": "sns:publish", "Effect": "Allow", "Resource": topic_arn}]},
110        }
111        return document
112
113    @classmethod
114    def event_bus_put_events_role_policy(cls, event_bus_arn, logical_id):
115        document = {
116            "PolicyName": logical_id + "EventBridgePolicy",
117            "PolicyDocument": {
118                "Statement": [{"Action": "events:PutEvents", "Effect": "Allow", "Resource": event_bus_arn}]
119            },
120        }
121        return document
122
123    @classmethod
124    def lambda_invoke_function_role_policy(cls, function_arn, logical_id):
125        document = {
126            "PolicyName": logical_id + "LambdaPolicy",
127            "PolicyDocument": {
128                "Statement": [{"Action": "lambda:InvokeFunction", "Effect": "Allow", "Resource": function_arn}]
129            },
130        }
131        return document
132