Allow SAM Function and SAM StateMachines to Propagate Tags to Generated Resources

.cfnlintrc.yaml

@@ -123,6 +123,7 @@ ignore_templates:
   - tests/translator/output/**/managed_policies_minimal.json  # Intentionally has non-existent managed policy name
   - tests/translator/output/**/function_with_mq.json  # Property "EventSourceArn" can Fn::GetAtt to a resource of types [AWS::DynamoDB::GlobalTable, AWS::DynamoDB::Table, AWS::Kinesis::Stream, AWS::Kinesis::StreamConsumer, AWS::SQS::Queue]
   - tests/translator/output/**/function_with_mq_using_autogen_role.json  # Property "EventSourceArn" can Fn::GetAtt to a resource of types [AWS::DynamoDB::GlobalTable, AWS::DynamoDB::Table, AWS::Kinesis::Stream, AWS::Kinesis::StreamConsumer, AWS::SQS::Queue]
+  - tests/translator/output/**/api_with_propagate_tags.json # Obsolete DependsOn on resource
 ignore_checks:
   - E2531 # Deprecated runtime; not relevant for transform tests
   - W2531 # EOL runtime; not relevant for transform tests

samtranslator/internal/schema_source/aws_serverless_function.py

@@ -618,6 +618,7 @@ class Properties(BaseModel):
     SnapStart: Optional[SnapStart] = prop("SnapStart")
     RuntimeManagementConfig: Optional[RuntimeManagementConfig] = prop("RuntimeManagementConfig")
     Tags: Optional[Tags] = prop("Tags")
+    PropagateTags: Optional[bool]  # TODO: add docs
     Timeout: Optional[Timeout] = prop("Timeout")
     Tracing: Optional[Tracing] = prop("Tracing")
     VersionDescription: Optional[PassThroughProp] = prop("VersionDescription")
@@ -647,6 +648,7 @@ class Globals(BaseModel):
         ["AWS::Lambda::Function", "Properties", "Environment"],
     )
     Tags: Optional[Tags] = prop("Tags")
+    PropagateTags: Optional[bool]  # TODO: add docs
     Tracing: Optional[Tracing] = prop("Tracing")
     KmsKeyArn: Optional[KmsKeyArn] = prop("KmsKeyArn")
     Layers: Optional[Layers] = prop("Layers")

samtranslator/internal/schema_source/aws_serverless_statemachine.py

@@ -169,6 +169,7 @@ class Properties(BaseModel):
     Role: Optional[PassThroughProp] = properties("Role")
     RolePath: Optional[PassThroughProp] = properties("RolePath")
     Tags: Optional[DictStrAny] = properties("Tags")
+    PropagateTags: Optional[bool]  # TODO: add docs
     Tracing: Optional[PassThroughProp] = properties("Tracing")
     Type: Optional[PassThroughProp] = properties("Type")
 
@@ -177,3 +178,7 @@ class Resource(ResourceAttributes):
     Type: Literal["AWS::Serverless::StateMachine"]
     Properties: Properties
     Connectors: Optional[Dict[str, EmbeddedConnector]]
+
+
+class Globals(BaseModel):
+    PropagateTags: Optional[bool]  # TODO: add docs

samtranslator/internal/schema_source/schema.py

@@ -27,6 +27,7 @@ class Globals(BaseModel):
     Api: Optional[aws_serverless_api.Globals]
     HttpApi: Optional[aws_serverless_httpapi.Globals]
     SimpleTable: Optional[aws_serverless_simpletable.Globals]
+    StateMachine: Optional[aws_serverless_statemachine.Globals]
 
 
 Resources = Union[

samtranslator/model/codedeploy.py

@@ -6,6 +6,7 @@ class CodeDeployApplication(Resource):
     resource_type = "AWS::CodeDeploy::Application"
     property_types = {
         "ComputePlatform": GeneratedProperty(),
+        "Tags": GeneratedProperty(),
     }
 
     runtime_attrs = {"name": lambda self: ref(self.logical_id)}
@@ -21,6 +22,7 @@ class CodeDeployDeploymentGroup(Resource):
         "DeploymentStyle": GeneratedProperty(),
         "ServiceRoleArn": GeneratedProperty(),
         "TriggerConfigurations": GeneratedProperty(),
+        "Tags": GeneratedProperty(),
     }
 
     runtime_attrs = {"name": lambda self: ref(self.logical_id)}

samtranslator/model/iot.py

@@ -6,6 +6,7 @@ class IotTopicRule(Resource):
     resource_type = "AWS::IoT::TopicRule"
     property_types = {
         "TopicRulePayload": GeneratedProperty(),
+        "Tags": GeneratedProperty(),
     }
 
     runtime_attrs = {"name": lambda self: ref(self.logical_id), "arn": lambda self: fnGetAtt(self.logical_id, "Arn")}

samtranslator/model/sam_resources.py

@@ -127,6 +127,7 @@ class SamFunction(SamResourceMacro):
         "Environment": PropertyType(False, dict_of(IS_STR, IS_DICT)),
         "Events": PropertyType(False, dict_of(IS_STR, IS_DICT)),
         "Tags": PropertyType(False, IS_DICT),
+        "PropagateTags": PropertyType(False, IS_BOOL),
         "Tracing": PropertyType(False, one_of(IS_DICT, IS_STR)),
         "KmsKeyArn": PassThroughProperty(False),
         "DeploymentPreference": PropertyType(False, IS_DICT),
@@ -169,6 +170,7 @@ class SamFunction(SamResourceMacro):
     Environment: Optional[Dict[str, Any]]
     Events: Optional[Dict[str, Any]]
     Tags: Optional[Dict[str, Any]]
+    PropagateTags: Optional[bool]
     Tracing: Optional[Dict[str, Any]]
     KmsKeyArn: Optional[Intrinsicable[str]]
     DeploymentPreference: Optional[Dict[str, Any]]
@@ -312,6 +314,8 @@ def to_cloudformation(self, **kwargs):  # type: ignore[no-untyped-def]
         except InvalidEventException as e:
             raise InvalidResourceException(self.logical_id, e.message) from e
 
+        self.propagate_tags(resources, self.Tags, self.PropagateTags)
+
         return resources
 
     def _construct_event_invoke_config(  # noqa: too-many-arguments
@@ -1716,6 +1720,7 @@ class SamStateMachine(SamResourceMacro):
         "Name": PropertyType(False, IS_STR),
         "Type": PropertyType(False, IS_STR),
         "Tags": PropertyType(False, IS_DICT),
+        "PropagateTags": PropertyType(False, IS_BOOL),
         "Policies": PropertyType(False, one_of(IS_STR, list_of(one_of(IS_STR, IS_DICT, IS_DICT)))),
         "Tracing": PropertyType(False, IS_DICT),
         "PermissionsBoundary": PropertyType(False, IS_STR),
@@ -1731,6 +1736,7 @@ class SamStateMachine(SamResourceMacro):
     Name: Optional[Intrinsicable[str]]
     Type: Optional[Intrinsicable[str]]
     Tags: Optional[Dict[str, Any]]
+    PropagateTags: Optional[bool]
     Policies: Optional[List[Any]]
     Tracing: Optional[Dict[str, Any]]
     PermissionsBoundary: Optional[Intrinsicable[str]]
@@ -1772,7 +1778,11 @@ def to_cloudformation(self, **kwargs):  # type: ignore[no-untyped-def]
             get_managed_policy_map=get_managed_policy_map,
         )
 
-        return state_machine_generator.to_cloudformation()
+        generated_resources = state_machine_generator.to_cloudformation()
+
+        self.propagate_tags(generated_resources, self.Tags, self.PropagateTags)
+
+        return generated_resources
 
     def resources_to_link(self, resources: Dict[str, Any]) -> Dict[str, Any]:
         try:

samtranslator/model/sns.py

@@ -27,5 +27,6 @@ class SNSTopic(Resource):
     resource_type = "AWS::SNS::Topic"
     property_types = {
         "TopicName": GeneratedProperty(),
+        "Tags": GeneratedProperty(),
     }
     runtime_attrs = {"arn": lambda self: ref(self.logical_id)}

samtranslator/model/sqs.py

@@ -6,7 +6,7 @@
 
 class SQSQueue(Resource):
     resource_type = "AWS::SQS::Queue"
-    property_types: Dict[str, PropertyType] = {}
+    property_types: Dict[str, PropertyType] = {"Tags": GeneratedProperty()}
     runtime_attrs = {
         "queue_url": lambda self: ref(self.logical_id),
         "arn": lambda self: fnGetAtt(self.logical_id, "Arn"),

samtranslator/plugins/api/implicit_api_plugin.py

@@ -93,6 +93,19 @@ def _generate_implicit_api_resource(self) -> Dict[str, Any]:
         Helper function implemented by child classes that create a new implicit API resource
         """
 
+    def _add_tags_to_implicit_api_if_necessary(self, resource: SamResource, template: SamTemplate) -> None:
+        """
+        Decides whether to add tags to the implicit api resource.
+        :param dict template_dict: SAM template dictionary
+        """
+        implicit_api_resource = template.get(self.IMPLICIT_API_LOGICAL_ID)
+        globals_var = template.get_globals().get(SamResourceType(resource.type).name, {})
+        should_propagate_tags = resource.properties.get("PropagateTags") or globals_var.get("PropagateTags")
+
+        if implicit_api_resource and resource.properties.get("Tags") and should_propagate_tags:
+            implicit_api_resource.properties.setdefault("Tags", {}).update(resource.properties["Tags"])
+            implicit_api_resource.properties["PropagateTags"] = True
+
     @cw_timer(prefix="Plugin-ImplicitApi")
     def on_before_transform_template(self, template_dict):  # type: ignore[no-untyped-def]
         """

samtranslator/plugins/api/implicit_http_api_plugin.py

@@ -60,6 +60,12 @@ def _process_api_events(  # noqa: too-many-arguments
             if not event_properties:
                 event["Properties"] = event_properties  # We are updating its Properties
 
+            # if the API ID is not provided in the event source properties, this implies that SAM-T will
+            # construct an implicit API resource for this API event source, and we need to add tags to the
+            # implicit API resource if customers specify `PropagateTags` property
+            if self.API_ID_EVENT_PROPERTY not in event_properties:
+                self._add_tags_to_implicit_api_if_necessary(function, template)
+
             self._add_implicit_api_id_if_necessary(event_properties)  # type: ignore[no-untyped-call]
 
             path = event_properties.get("Path", "")

samtranslator/plugins/api/implicit_rest_api_plugin.py

@@ -61,6 +61,12 @@ def _process_api_events(  # noqa: too-many-arguments
 
             sam_expect(event_properties, event_id, "", is_sam_event=True).to_be_a_map("Properties should be a map.")
 
+            # if the API ID is not provided in the event source properties, this implies that SAM-T will
+            # construct an implicit API resource for this API event source, and we need to add tags to the
+            # implicit API resource if customers specify `PropagateTags` property
+            if self.API_ID_EVENT_PROPERTY not in event_properties:
+                self._add_tags_to_implicit_api_if_necessary(function, template)
+
             self._add_implicit_api_id_if_necessary(event_properties)  # type: ignore[no-untyped-call]
 
             api_id, path, method = self._validate_api_event(event_id, event_properties)

samtranslator/plugins/globals/globals.py

@@ -32,6 +32,7 @@ class Globals:
             "VpcConfig",
             "Environment",
             "Tags",
+            "PropagateTags",
             "Tracing",
             "KmsKeyArn",
             "AutoPublishAlias",
@@ -91,6 +92,7 @@ class Globals:
             "PropagateTags",
         ],
         SamResourceType.SimpleTable.value: ["SSESpecification"],
+        SamResourceType.StateMachine.value: ["PropagateTags"],
     }
     # unreleased_properties *must be* part of supported_properties too
     unreleased_properties: Dict[str, List[str]] = {

samtranslator/schema/schema.json

@@ -239150,6 +239150,9 @@
         },
         "SimpleTable": {
           "$ref": "#/definitions/samtranslator__internal__schema_source__aws_serverless_simpletable__Globals"
+        },
+        "StateMachine": {
+          "$ref": "#/definitions/samtranslator__internal__schema_source__aws_serverless_statemachine__Globals"
         }
       },
       "title": "Globals",
@@ -240582,6 +240585,10 @@
           "title": "PermissionsBoundary",
           "type": "string"
         },
+        "PropagateTags": {
+          "title": "Propagatetags",
+          "type": "boolean"
+        },
         "ProvisionedConcurrencyConfig": {
           "allOf": [
             {
@@ -240956,6 +240963,10 @@
           "markdownDescription": "Permission policies for this function\\. Policies will be appended to the function's default AWS Identity and Access Management \\(IAM\\) execution role\\.  \nThis property accepts a single value or list of values\\. Allowed values include:  \n+ [AWS SAM policy templates](https://docs.aws.amazon.com/serverless-application-model/latest/developerguide/serverless-policy-templates.html)\\.\n+ The ARN of an [AWS managed policy](https://docs.aws.amazon.com/IAM/latest/UserGuide/access_policies_managed-vs-inline.html#aws-managed-policies) or [ customer managed policy](https://docs.aws.amazon.com/IAM/latest/UserGuide/access_policies_managed-vs-inline.html#customer-managed-policies)\\.\n+ The name of an AWS managed policy from the following [ list](https://github.com/aws/serverless-application-model/blob/develop/samtranslator/internal/data/aws_managed_policies.json)\\.\n+ An [ inline IAM policy](https://docs.aws.amazon.com/IAM/latest/UserGuide/access_policies_managed-vs-inline.html#inline-policies) formatted in YAML as a map\\.\nIf you set the `Role` property, this property is ignored\\.\n*Type*: String \\| List \\| Map  \n*Required*: No  \n*AWS CloudFormation compatibility*: This property is similar to the [`Policies`](https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-iam-role.html#cfn-iam-role-policies) property of an `AWS::IAM::Role` resource\\.",
           "title": "Policies"
         },
+        "PropagateTags": {
+          "title": "Propagatetags",
+          "type": "boolean"
+        },
         "ProvisionedConcurrencyConfig": {
           "$ref": "#/definitions/AWS::Lambda::Alias.ProvisionedConcurrencyConfiguration",
           "markdownDescription": "The provisioned concurrency configuration of a function's alias\\.  \n`ProvisionedConcurrencyConfig` can be specified only if the `AutoPublishAlias` is set\\. Otherwise, an error results\\.\n*Type*: [ProvisionedConcurrencyConfig](https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-lambda-alias.html#cfn-lambda-alias-provisionedconcurrencyconfig)  \n*Required*: No  \n*AWS CloudFormation compatibility*: This property is passed directly to the [`ProvisionedConcurrencyConfig`](https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-lambda-alias.html#cfn-lambda-alias-provisionedconcurrencyconfig) property of an `AWS::Lambda::Alias` resource\\.",
@@ -242478,6 +242489,17 @@
       "title": "EventBridgeRuleTarget",
       "type": "object"
     },
+    "samtranslator__internal__schema_source__aws_serverless_statemachine__Globals": {
+      "additionalProperties": false,
+      "properties": {
+        "PropagateTags": {
+          "title": "Propagatetags",
+          "type": "boolean"
+        }
+      },
+      "title": "Globals",
+      "type": "object"
+    },
     "samtranslator__internal__schema_source__aws_serverless_statemachine__Properties": {
       "additionalProperties": false,
       "properties": {
@@ -242579,6 +242601,10 @@
           "markdownDescription": "Permission policies for this state machine\\. Policies will be appended to the state machine's default AWS Identity and Access Management \\(IAM\\) execution role\\.  \nThis property accepts a single value or list of values\\. Allowed values include:  \n+ [AWS SAM policy templates](https://docs.aws.amazon.com/serverless-application-model/latest/developerguide/serverless-policy-templates.html)\\.\n+ The ARN of an [AWS managed policy](https://docs.aws.amazon.com/IAM/latest/UserGuide/access_policies_managed-vs-inline.html#aws-managed-policies) or [customer managed policy](https://docs.aws.amazon.com/IAM/latest/UserGuide/access_policies_managed-vs-inline.html#customer-managed-policies)\\.\n+ The name of an AWS managed policy from the following [ list](https://github.com/aws/serverless-application-model/blob/develop/samtranslator/internal/data/aws_managed_policies.json)\\.\n+ An [ inline IAM policy](https://docs.aws.amazon.com/IAM/latest/UserGuide/access_policies_managed-vs-inline.html#inline-policies) formatted in YAML as a map\\.\nIf you set the `Role` property, this property is ignored\\.\n*Type*: String \\| List \\| Map  \n*Required*: No  \n*AWS CloudFormation compatibility*: This property is unique to AWS SAM and doesn't have an AWS CloudFormation equivalent\\.",
           "title": "Policies"
         },
+        "PropagateTags": {
+          "title": "Propagatetags",
+          "type": "boolean"
+        },
         "Role": {
           "allOf": [
             {

samtranslator/sdk/template.py

@@ -55,6 +55,14 @@ def set(  # noqa: builtin-attribute-shadowing
 
         self.resources[logical_id] = resource_dict
 
+    def get_globals(self) -> Dict[str, Any]:
+        """
+        Gets the global section of the template
+
+        :return dict: Global section of the template
+        """
+        return self.template_dict.get("Globals") or {}
+
     def get(self, logical_id: str) -> Optional[SamResource]:
         """
         Gets the resource at the given logical_id if present

samtranslator/translator/translator.py

@@ -154,6 +154,7 @@ def translate(  # noqa: too-many-branches
         mappings_resolver = IntrinsicsResolver(
             template.get("Mappings", {}), {FindInMapAction.intrinsic_name: FindInMapAction()}
         )
+
         deployment_preference_collection = DeploymentPreferenceCollection()
         supported_resource_refs = SupportedResourceReferences()
         shared_api_usage_plan = SharedApiUsagePlan()