feat: Add Step Function Resource (#1601)

Co-authored-by: Jacob Fuss <jfuss@users.noreply.github.com>

samtranslator/__init__.py

@@ -1 +1 @@
-__version__ = "1.23.0"
+__version__ = "1.24.0"

samtranslator/model/iam.py

@@ -18,6 +18,36 @@ class IAMRole(Resource):
 
 
 class IAMRolePolicies:
+    @classmethod
+    def construct_assume_role_policy_for_service_principal(cls, service_principal):
+        document = {
+            "Version": "2012-10-17",
+            "Statement": [
+                {"Action": ["sts:AssumeRole"], "Effect": "Allow", "Principal": {"Service": [service_principal]},}
+            ],
+        }
+        return document
+
+    @classmethod
+    def step_functions_start_execution_role_policy(cls, state_machine_arn, logical_id):
+        document = {
+            "PolicyName": logical_id + "StartExecutionPolicy",
+            "PolicyDocument": {
+                "Statement": [{"Action": "states:StartExecution", "Effect": "Allow", "Resource": state_machine_arn}]
+            },
+        }
+        return document
+
+    @classmethod
+    def stepfunctions_assume_role_policy(cls):
+        document = {
+            "Version": "2012-10-17",
+            "Statement": [
+                {"Action": ["sts:AssumeRole"], "Effect": "Allow", "Principal": {"Service": ["states.amazonaws.com"]},}
+            ],
+        }
+        return document
+
     @classmethod
     def cloud_watch_log_assume_role_policy(cls):
         document = {

samtranslator/model/resource_policies.py

@@ -0,0 +1,200 @@
+from enum import Enum
+from collections import namedtuple
+
+from six import string_types
+
+from samtranslator.model.intrinsics import is_intrinsic, is_intrinsic_if, is_intrinsic_no_value
+from samtranslator.model.exceptions import InvalidTemplateException
+
+PolicyEntry = namedtuple("PolicyEntry", "data type")
+
+
+class ResourcePolicies(object):
+    """
+    Class encapsulating the policies property of SAM resources. This class strictly encapsulates the data
+    and does not take opinions on how to handle them.
+
+    There are three types of policies:
+        - Policy Statements
+        - AWS or Custom Managed Policy names/arns
+        - Policy Templates
+
+    This class is capable of parsing and detecting the type of the policy. Optionally, if policy template information
+    is provided to this class, it will detect Policy Templates too.
+    """
+
+    POLICIES_PROPERTY_NAME = "Policies"
+
+    def __init__(self, resource_properties, policy_template_processor=None):
+        """
+        Initialize with policies data from resource's properties
+
+        :param dict resource_properties: Dictionary containing properties of this resource
+        :param policy_template_processor: Optional Instance of PolicyTemplateProcessor that can conclusively detect
+            if a given policy is a template or not. If not provided, then this class will not detect policy templates.
+        """
+
+        # This variable is required to get policies
+        self._policy_template_processor = policy_template_processor
+
+        # Build the list of policies upon construction.
+        self.policies = self._get_policies(resource_properties)
+
+    def get(self):
+        """
+        Iterator method that "yields" the next policy entry on subsequent calls to this method.
+
+        :yields namedtuple("data", "type"): Yields a named tuple containing the policy data and its type
+        """
+
+        for policy_tuple in self.policies:
+            yield policy_tuple
+
+    def __len__(self):
+        return len(self.policies)
+
+    def _get_policies(self, resource_properties):
+        """
+        Returns a list of policies from the resource properties. This method knows how to interpret and handle
+        polymorphic nature of the policies property.
+
+        Policies can be one of the following:
+
+            * Managed policy name: string
+            * List of managed policy names: list of strings
+            * IAM Policy document: dict containing Statement key
+            * List of IAM Policy documents: list of IAM Policy Document
+            * Policy Template: dict with only one key where key is in list of supported policy template names
+            * List of Policy Templates: list of Policy Template
+
+
+        :param dict resource_properties: Dictionary of resource properties containing the policies property.
+            It is assumed that this is already a dictionary and contains policies key.
+        :return list of PolicyEntry: List of policies, where each item is an instance of named tuple `PolicyEntry`
+        """
+
+        policies = None
+
+        if self._contains_policies(resource_properties):
+            policies = resource_properties[self.POLICIES_PROPERTY_NAME]
+
+        if not policies:
+            # Policies is None or empty
+            return []
+
+        if not isinstance(policies, list):
+            # Just a single entry. Make it into a list of convenience
+            policies = [policies]
+
+        result = []
+        for policy in policies:
+            policy_type = self._get_type(policy)
+            entry = PolicyEntry(data=policy, type=policy_type)
+            result.append(entry)
+
+        return result
+
+    def _contains_policies(self, resource_properties):
+        """
+        Is there policies data in this resource?
+
+        :param dict resource_properties: Properties of the resource
+        :return: True if we can process this resource. False, otherwise
+        """
+        return (
+            resource_properties is not None
+            and isinstance(resource_properties, dict)
+            and self.POLICIES_PROPERTY_NAME in resource_properties
+        )
+
+    def _get_type(self, policy):
+        """
+        Returns the type of the given policy
+
+        :param string or dict policy: Policy data
+        :return PolicyTypes: Type of the given policy. None, if type could not be inferred
+        """
+
+        # Must handle intrinsic functions. Policy could be a primitive type or an intrinsic function
+
+        # Managed policies are of type string
+        if isinstance(policy, string_types):
+            return PolicyTypes.MANAGED_POLICY
+
+        # Handle the special case for 'if' intrinsic function
+        if is_intrinsic_if(policy):
+            return self._get_type_from_intrinsic_if(policy)
+
+        # Intrinsic functions are treated as managed policies by default
+        if is_intrinsic(policy):
+            return PolicyTypes.MANAGED_POLICY
+
+        # Policy statement is a dictionary with the key "Statement" in it
+        if isinstance(policy, dict) and "Statement" in policy:
+            return PolicyTypes.POLICY_STATEMENT
+
+        # This could be a policy template then.
+        if self._is_policy_template(policy):
+            return PolicyTypes.POLICY_TEMPLATE
+
+        # Nothing matches. Don't take opinions on how to handle it. Instead just set the appropriate type.
+        return PolicyTypes.UNKNOWN
+
+    def _is_policy_template(self, policy):
+        """
+        Is the given policy data a policy template? Policy templates is a dictionary with one key which is the name
+        of the template.
+
+        :param dict policy: Policy data
+        :return: True, if this is a policy template. False if it is not
+        """
+
+        return (
+            self._policy_template_processor is not None
+            and isinstance(policy, dict)
+            and len(policy) == 1
+            and self._policy_template_processor.has(list(policy.keys())[0]) is True
+        )
+
+    def _get_type_from_intrinsic_if(self, policy):
+        """
+        Returns the type of the given policy assuming that it is an intrinsic if function
+
+        :param policy: Input value to get type from
+        :return: PolicyTypes: Type of the given policy. PolicyTypes.UNKNOWN, if type could not be inferred
+        """
+        intrinsic_if_value = policy["Fn::If"]
+
+        if not len(intrinsic_if_value) == 3:
+            raise InvalidTemplateException("Fn::If requires 3 arguments")
+
+        if_data = intrinsic_if_value[1]
+        else_data = intrinsic_if_value[2]
+
+        if_data_type = self._get_type(if_data)
+        else_data_type = self._get_type(else_data)
+
+        if if_data_type == else_data_type:
+            return if_data_type
+
+        if is_intrinsic_no_value(if_data):
+            return else_data_type
+
+        if is_intrinsic_no_value(else_data):
+            return if_data_type
+
+        raise InvalidTemplateException(
+            "Different policy types within the same Fn::If statement is unsupported. "
+            "Separate different policy types into different Fn::If statements"
+        )
+
+
+class PolicyTypes(Enum):
+    """
+    Enum of different policy types supported by SAM & this plugin
+    """
+
+    MANAGED_POLICY = "managed_policy"
+    POLICY_STATEMENT = "policy_statement"
+    POLICY_TEMPLATE = "policy_template"
+    UNKNOWN = "unknown"

samtranslator/model/role_utils/__init__.py

@@ -0,0 +1 @@
+from .role_constructor import construct_role_for_resource

samtranslator/model/role_utils/role_constructor.py

@@ -0,0 +1,111 @@
+from six import string_types
+
+from samtranslator.model.iam import IAMRole
+from samtranslator.model.resource_policies import ResourcePolicies, PolicyTypes
+from samtranslator.model.intrinsics import is_intrinsic_if, is_intrinsic_no_value
+from samtranslator.model.exceptions import InvalidResourceException
+
+
+def construct_role_for_resource(
+    resource_logical_id,
+    attributes,
+    managed_policy_map,
+    assume_role_policy_document,
+    resource_policies,
+    managed_policy_arns=None,
+    policy_documents=None,
+    permissions_boundary=None,
+    tags=None,
+):
+    """
+    Constructs an execution role for a resource.
+    :param resource_logical_id: The logical_id of the SAM resource that the role will be associated with
+    :param attributes: Map of resource attributes to their values
+    :param managed_policy_map: Map of managed policy names to the ARNs
+    :param assume_role_policy_document: The trust policy that must be associated with the role
+    :param resource_policies: ResourcePolicies object encapuslating the policies property of SAM resource
+    :param managed_policy_arns: List of managed policy ARNs to be associated with the role
+    :param policy_documents: List of policy documents to be associated with the role
+    :param permissions_boundary: The ARN of the policy used to set the permissions boundary for the role
+    :param tags: Tags to be associated with the role
+
+    :returns: the generated IAM Role
+    :rtype: model.iam.IAMRole
+    """
+    role_logical_id = resource_logical_id + "Role"
+    execution_role = IAMRole(logical_id=role_logical_id, attributes=attributes)
+    execution_role.AssumeRolePolicyDocument = assume_role_policy_document
+
+    if not managed_policy_arns:
+        managed_policy_arns = []
+
+    if not policy_documents:
+        policy_documents = []
+
+    for index, policy_entry in enumerate(resource_policies.get()):
+        if policy_entry.type is PolicyTypes.POLICY_STATEMENT:
+
+            if is_intrinsic_if(policy_entry.data):
+
+                intrinsic_if = policy_entry.data
+                then_statement = intrinsic_if["Fn::If"][1]
+                else_statement = intrinsic_if["Fn::If"][2]
+
+                if not is_intrinsic_no_value(then_statement):
+                    then_statement = {
+                        "PolicyName": execution_role.logical_id + "Policy" + str(index),
+                        "PolicyDocument": then_statement,
+                    }
+                    intrinsic_if["Fn::If"][1] = then_statement
+
+                if not is_intrinsic_no_value(else_statement):
+                    else_statement = {
+                        "PolicyName": execution_role.logical_id + "Policy" + str(index),
+                        "PolicyDocument": else_statement,
+                    }
+                    intrinsic_if["Fn::If"][2] = else_statement
+
+                policy_documents.append(intrinsic_if)
+
+            else:
+                policy_documents.append(
+                    {
+                        "PolicyName": execution_role.logical_id + "Policy" + str(index),
+                        "PolicyDocument": policy_entry.data,
+                    }
+                )
+
+        elif policy_entry.type is PolicyTypes.MANAGED_POLICY:
+
+            # There are three options:
+            #   Managed Policy Name (string): Try to convert to Managed Policy ARN
+            #   Managed Policy Arn (string): Insert it directly into the list
+            #   Intrinsic Function (dict): Insert it directly into the list
+            #
+            # When you insert into managed_policy_arns list, de-dupe to prevent same ARN from showing up twice
+            #
+
+            policy_arn = policy_entry.data
+            if isinstance(policy_entry.data, string_types) and policy_entry.data in managed_policy_map:
+                policy_arn = managed_policy_map[policy_entry.data]
+
+            # De-Duplicate managed policy arns before inserting. Mainly useful
+            # when customer specifies a managed policy which is already inserted
+            # by SAM, such as AWSLambdaBasicExecutionRole
+            if policy_arn not in managed_policy_arns:
+                managed_policy_arns.append(policy_arn)
+        else:
+            # Policy Templates are not supported here in the "core"
+            raise InvalidResourceException(
+                resource_logical_id,
+                "Policy at index {} in the '{}' property is not valid".format(
+                    index, resource_policies.POLICIES_PROPERTY_NAME
+                ),
+            )
+
+    execution_role.ManagedPolicyArns = list(managed_policy_arns)
+    execution_role.Policies = policy_documents or None
+    execution_role.PermissionsBoundary = permissions_boundary
+    execution_role.Tags = tags
+
+    return execution_role