Release 1.79.0 (to main)

samtranslator/__init__.py

@@ -1 +1 @@
-__version__ = "1.78.0"
+__version__ = "1.79.0"

samtranslator/internal/schema_source/aws_serverless_statemachine.py

@@ -49,6 +49,7 @@ class ScheduleEventProperties(BaseModel):
     Schedule: Optional[PassThroughProp] = scheduleeventproperties("Schedule")
     State: Optional[PassThroughProp] = scheduleeventproperties("State")
     Target: Optional[ScheduleTarget] = scheduleeventproperties("Target")
+    RoleArn: Optional[PassThroughProp]  # TODO: add doc
 
 
 class ScheduleEvent(BaseModel):

samtranslator/model/eventsources/push.py

@@ -12,7 +12,15 @@
 from samtranslator.model.eventsources import FUNCTION_EVETSOURCE_METRIC_PREFIX
 from samtranslator.model.eventsources.pull import SQS
 from samtranslator.model.exceptions import InvalidDocumentException, InvalidEventException, InvalidResourceException
-from samtranslator.model.intrinsics import fnGetAtt, fnSub, is_intrinsic, make_conditional, make_shorthand, ref
+from samtranslator.model.intrinsics import (
+    fnGetAtt,
+    fnSub,
+    get_logical_id_from_intrinsic,
+    is_intrinsic,
+    make_conditional,
+    make_shorthand,
+    ref,
+)
 from samtranslator.model.iot import IotTopicRule
 from samtranslator.model.lambda_ import LambdaPermission
 from samtranslator.model.s3 import S3Bucket
@@ -517,6 +525,8 @@ def to_cloudformation(self, **kwargs):  # type: ignore[no-untyped-def]
         if not function:
             raise TypeError("Missing required keyword argument: function")
 
+        intrinsics_resolver: IntrinsicsResolver = kwargs["intrinsics_resolver"]
+
         # SNS -> Lambda
         if not self.SqsSubscription:
             subscription = self._inject_subscription(
@@ -534,7 +544,11 @@ def to_cloudformation(self, **kwargs):  # type: ignore[no-untyped-def]
         # SNS -> SQS(Create New) -> Lambda
         if isinstance(self.SqsSubscription, bool):
             resources = []  # type: ignore[var-annotated]
-            queue = self._inject_sqs_queue(function)  # type: ignore[no-untyped-call]
+
+            fifo_topic = self._check_fifo_topic(
+                get_logical_id_from_intrinsic(self.Topic), kwargs.get("original_template"), intrinsics_resolver
+            )
+            queue = self._inject_sqs_queue(function, fifo_topic)  # type: ignore[no-untyped-call]
             queue_arn = queue.get_runtime_attr("arn")
             queue_url = queue.get_runtime_attr("queue_url")
 
@@ -591,6 +605,19 @@ def to_cloudformation(self, **kwargs):  # type: ignore[no-untyped-def]
         resources.append(subscription)
         return resources
 
+    def _check_fifo_topic(
+        self,
+        topic_id: Optional[str],
+        template: Optional[Dict[str, Any]],
+        intrinsics_resolver: IntrinsicsResolver,
+    ) -> bool:
+        if not topic_id or not template:
+            return False
+
+        resources = template.get("Resources", {})
+        properties = resources.get(topic_id, {}).get("Properties", {})
+        return intrinsics_resolver.resolve_parameter_refs(properties.get("FifoTopic", False))  # type: ignore[no-any-return]
+
     def _inject_subscription(  # noqa: PLR0913
         self,
         protocol: str,
@@ -621,8 +648,12 @@ def _inject_subscription(  # noqa: PLR0913
 
         return subscription
 
-    def _inject_sqs_queue(self, function):  # type: ignore[no-untyped-def]
-        return SQSQueue(self.logical_id + "Queue", attributes=function.get_passthrough_resource_attributes())
+    def _inject_sqs_queue(self, function, fifo_topic=False):  # type: ignore[no-untyped-def]
+        queue = SQSQueue(self.logical_id + "Queue", attributes=function.get_passthrough_resource_attributes())
+
+        if fifo_topic:
+            queue.FifoQueue = fifo_topic
+        return queue
 
     def _inject_sqs_event_source_mapping(self, function, role, queue_arn, batch_size=None, enabled=None):  # type: ignore[no-untyped-def]
         event_source = SQS(

samtranslator/model/sam_resources.py

@@ -351,6 +351,7 @@ def to_cloudformation(self, **kwargs):  # type: ignore[no-untyped-def] # noqa: P
                 kwargs["event_resources"],
                 intrinsics_resolver,
                 lambda_alias=lambda_alias,
+                original_template=kwargs.get("original_template"),
             )
         except InvalidEventException as e:
             raise InvalidResourceException(self.logical_id, e.message) from e
@@ -775,13 +776,14 @@ def order_events(event: Tuple[str, Any]) -> Any:
             return logical_id
         return event_dict.get("Properties", {}).get("Path", logical_id)
 
-    def _generate_event_resources(
+    def _generate_event_resources(  # noqa: PLR0913
         self,
         lambda_function: LambdaFunction,
         execution_role: Optional[IAMRole],
         event_resources: Any,
         intrinsics_resolver: IntrinsicsResolver,
         lambda_alias: Optional[LambdaAlias] = None,
+        original_template: Optional[Dict[str, Any]] = None,
     ) -> List[Any]:
         """Generates and returns the resources associated with this function's events.
 
@@ -811,6 +813,7 @@ def _generate_event_resources(
                     "function": lambda_alias or lambda_function,
                     "role": execution_role,
                     "intrinsics_resolver": intrinsics_resolver,
+                    "original_template": original_template,
                 }
 
                 for name, resource in event_resources[logical_id].items():

samtranslator/model/sqs.py

@@ -2,16 +2,22 @@
 
 from samtranslator.model import GeneratedProperty, PropertyType, Resource
 from samtranslator.model.intrinsics import fnGetAtt, ref
+from samtranslator.model.types import PassThrough
 
 
 class SQSQueue(Resource):
     resource_type = "AWS::SQS::Queue"
-    property_types: Dict[str, PropertyType] = {"Tags": GeneratedProperty()}
+    property_types: Dict[str, PropertyType] = {
+        "FifoQueue": GeneratedProperty(),
+        "Tags": GeneratedProperty(),
+    }
     runtime_attrs = {
         "queue_url": lambda self: ref(self.logical_id),
         "arn": lambda self: fnGetAtt(self.logical_id, "Arn"),
     }
 
+    FifoQueue: PassThrough
+
 
 class SQSQueuePolicy(Resource):
     resource_type = "AWS::SQS::QueuePolicy"

samtranslator/model/stepfunctions/events.py

@@ -1,6 +1,6 @@
 import json
 from abc import ABCMeta
-from typing import Any, Dict, Optional, cast
+from typing import Any, Dict, List, Optional, Union, cast
 
 from samtranslator.metrics.method_decorator import cw_timer
 from samtranslator.model import Property, PropertyType, Resource, ResourceMacro
@@ -10,6 +10,7 @@
 from samtranslator.model.exceptions import InvalidEventException
 from samtranslator.model.iam import IAMRole, IAMRolePolicies
 from samtranslator.model.intrinsics import fnSub
+from samtranslator.model.stepfunctions.resources import StepFunctionsStateMachine
 from samtranslator.model.types import IS_BOOL, IS_DICT, IS_STR, PassThrough
 from samtranslator.swagger.swagger import SwaggerEditor
 from samtranslator.translator import logical_id_generator
@@ -50,7 +51,13 @@ def _generate_logical_id(self, prefix, suffix, resource_type):  # type: ignore[n
         generator = logical_id_generator.LogicalIdGenerator(prefix + resource_type, suffix)
         return generator.gen()
 
-    def _construct_role(self, resource, permissions_boundary=None, prefix=None, suffix=""):  # type: ignore[no-untyped-def]
+    def _construct_role(
+        self,
+        resource: StepFunctionsStateMachine,
+        permissions_boundary: Optional[str],
+        prefix: Optional[str],
+        suffix: str = "",
+    ) -> IAMRole:
         """Constructs the IAM Role resource allowing the event service to invoke
         the StartExecution API of the state machine resource it is associated with.
 
@@ -93,6 +100,7 @@ class Schedule(EventSource):
         "DeadLetterConfig": PropertyType(False, IS_DICT),
         "RetryPolicy": PropertyType(False, IS_DICT),
         "Target": Property(False, IS_DICT),
+        "RoleArn": Property(False, IS_STR),
     }
 
     Schedule: PassThrough
@@ -104,6 +112,7 @@ class Schedule(EventSource):
     DeadLetterConfig: Optional[Dict[str, Any]]
     RetryPolicy: Optional[PassThrough]
     Target: Optional[PassThrough]
+    RoleArn: Optional[PassThrough]
 
     @cw_timer(prefix=SFN_EVETSOURCE_METRIC_PREFIX)
     def to_cloudformation(self, resource, **kwargs):  # type: ignore[no-untyped-def]
@@ -113,7 +122,7 @@ def to_cloudformation(self, resource, **kwargs):  # type: ignore[no-untyped-def]
         :returns: a list of vanilla CloudFormation Resources, to which this Schedule event expands
         :rtype: list
         """
-        resources = []
+        resources: List[Any] = []
 
         permissions_boundary = kwargs.get("permissions_boundary")
 
@@ -135,8 +144,12 @@ def to_cloudformation(self, resource, **kwargs):  # type: ignore[no-untyped-def]
         events_rule.Name = self.Name
         events_rule.Description = self.Description
 
-        role = self._construct_role(resource, permissions_boundary)  # type: ignore[no-untyped-call]
-        resources.append(role)
+        role: Union[IAMRole, str, Dict[str, Any]]
+        if self.RoleArn is None:
+            role = self._construct_role(resource, permissions_boundary, prefix=None)
+            resources.append(role)
+        else:
+            role = self.RoleArn
 
         source_arn = events_rule.get_runtime_attr("arn")
         dlq_queue_arn = None
@@ -146,26 +159,44 @@ def to_cloudformation(self, resource, **kwargs):  # type: ignore[no-untyped-def]
                 self, source_arn, passthrough_resource_attributes
             )
             resources.extend(dlq_resources)
-        events_rule.Targets = [self._construct_target(resource, role, dlq_queue_arn)]  # type: ignore[no-untyped-call]
+        events_rule.Targets = [self._construct_target(resource, role, dlq_queue_arn)]
 
         return resources
 
-    def _construct_target(self, resource, role, dead_letter_queue_arn=None):  # type: ignore[no-untyped-def]
-        """Constructs the Target property for the EventBridge Rule.
-
-        :returns: the Target property
-        :rtype: dict
+    def _construct_target(
+        self,
+        resource: StepFunctionsStateMachine,
+        role: Union[IAMRole, str, Dict[str, Any]],
+        dead_letter_queue_arn: Optional[str],
+    ) -> Dict[str, Any]:
+        """_summary_
+
+        Parameters
+        ----------
+        resource
+            StepFunctionsState machine resource to be generated
+        role
+            The role to be used by the Schedule event resource either generated or user provides arn
+        dead_letter_queue_arn
+            Dead letter queue associated with the resource
+
+        Returns
+        -------
+            The Target property
         """
         target_id = (
             self.Target["Id"]
             if self.Target and "Id" in self.Target
             else generate_valid_target_id(self.logical_id, EVENT_RULE_SFN_TARGET_SUFFIX)
         )
+
         target = {
             "Arn": resource.get_runtime_attr("arn"),
             "Id": target_id,
-            "RoleArn": role.get_runtime_attr("arn"),
         }
+
+        target["RoleArn"] = role.get_runtime_attr("arn") if isinstance(role, IAMRole) else role
+
         if self.Input is not None:
             target["Input"] = self.Input
 
@@ -216,7 +247,7 @@ def to_cloudformation(self, resource, **kwargs):  # type: ignore[no-untyped-def]
         :returns: a list of vanilla CloudFormation Resources, to which this CloudWatch Events/EventBridge event expands
         :rtype: list
         """
-        resources = []
+        resources: List[Any] = []
 
         permissions_boundary = kwargs.get("permissions_boundary")
 
@@ -231,7 +262,11 @@ def to_cloudformation(self, resource, **kwargs):  # type: ignore[no-untyped-def]
 
         resources.append(events_rule)
 
-        role = self._construct_role(resource, permissions_boundary)  # type: ignore[no-untyped-call]
+        role = self._construct_role(
+            resource,
+            permissions_boundary,
+            prefix=None,
+        )
         resources.append(role)
 
         source_arn = events_rule.get_runtime_attr("arn")
@@ -331,7 +366,7 @@ def to_cloudformation(self, resource, **kwargs):  # type: ignore[no-untyped-def]
         :returns: a list of vanilla CloudFormation Resources, to which this Api event expands
         :rtype: list
         """
-        resources = []
+        resources: List[Any] = []
 
         intrinsics_resolver = kwargs.get("intrinsics_resolver")
         permissions_boundary = kwargs.get("permissions_boundary")
@@ -340,7 +375,7 @@ def to_cloudformation(self, resource, **kwargs):  # type: ignore[no-untyped-def]
             # Convert to lower case so that user can specify either GET or get
             self.Method = self.Method.lower()
 
-        role = self._construct_role(resource, permissions_boundary)  # type: ignore[no-untyped-call]
+        role = self._construct_role(resource, permissions_boundary, prefix=None)
         resources.append(role)
 
         explicit_api = kwargs["explicit_api"]

samtranslator/plugins/application/serverless_app_plugin.py

@@ -1,6 +1,7 @@
 import copy
 import json
 import logging
+import re
 from time import sleep
 from typing import Any, Callable, Dict, List, Optional, Tuple
 
@@ -150,6 +151,14 @@ def on_before_transform_template(self, template_dict):  # type: ignore[no-untype
                         raise InvalidResourceException(
                             logical_id, "Serverless Application Repository is not available in this region."
                         )
+                    # SSM Pattern found here https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/dynamic-references.html
+                    ssm_pattern = r"{{resolve:ssm:[a-zA-Z0-9_.\-/]+(:\d+)?}}"
+                    if re.search(ssm_pattern, app_id):
+                        raise InvalidResourceException(
+                            logical_id,
+                            "Serverless Application Repostiory does not support dynamic reference in 'ApplicationId' property.",
+                        )
+
                     self._make_service_call_with_retry(service_call, app_id, semver, key, logical_id)  # type: ignore[no-untyped-call]
                 except InvalidResourceException as e:
                     # Catch all InvalidResourceExceptions, raise those in the before_resource_transform target.