fix: Fail on generated IAM role logical ID collisions

samtranslator/model/exceptions.py

@@ -65,13 +65,23 @@ class DuplicateLogicalIdException(ExceptionWithMessage):
         message -- explanation of the error
     """
 
-    def __init__(self, logical_id: str, duplicate_id: str, resource_type: str) -> None:
+    def __init__(
+        self, logical_id: str, duplicate_id: str, resource_type: str, conflicting_logical_id: str | None = None
+    ) -> None:
         self._logical_id = logical_id
         self._duplicate_id = duplicate_id
         self._type = resource_type
+        self._conflicting_logical_id = conflicting_logical_id
 
     @property
     def message(self) -> str:
+        if self._conflicting_logical_id:
+            return (
+                f"Transforming resource with id [{self._logical_id}] attempts to create a new"
+                f' resource with id [{self._duplicate_id}] and type "{self._type}". Resource with id'
+                f" [{self._conflicting_logical_id}] already generates a resource with that id."
+                " Please use a different id for one of these resources."
+            )
         return (
             f"Transforming resource with id [{self._logical_id}] attempts to create a new"
             f' resource with id [{self._duplicate_id}] and type "{self._type}". A resource with that id already'

samtranslator/translator/translator.py

@@ -36,11 +36,15 @@
 from samtranslator.policy_template_processor.processor import PolicyTemplatesProcessor
 from samtranslator.sdk.parameter import SamParameterValues
 from samtranslator.translator.arn_generator import ArnGenerator
-from samtranslator.translator.verify_logical_id import verify_unique_logical_id
+from samtranslator.translator.verify_logical_id import GeneratedLogicalIdTracker, verify_unique_logical_id
 from samtranslator.utils.actions import ResolveDependsOn
 from samtranslator.utils.traverse import traverse
 from samtranslator.validator.value_validator import sam_expect
 
+# Source recorded for the CodeDeploy service role in GeneratedLogicalIdTracker. It contains "::", so it can never be
+# equal to a SAM resource logical id.
+CODEDEPLOY_SERVICE_ROLE_SOURCE = "AWS::CodeDeploy::ServiceRole"
+
 
 class Translator:
     """Translates SAM templates into CloudFormation templates"""
@@ -162,6 +166,7 @@ def translate(  # noqa: PLR0912, PLR0915
         shared_api_usage_plan = SharedApiUsagePlan()
         changed_logical_ids = {}
         route53_record_set_groups: dict[Any, Any] = {}
+        generated_logical_ids = GeneratedLogicalIdTracker()
         for logical_id, resource_dict in self._get_resources_to_iterate(sam_template, macro_resolver):
             try:
                 macro = macro_resolver.resolve_resource_type(resource_dict).from_dict(
@@ -194,7 +199,14 @@ def translate(  # noqa: PLR0912, PLR0915
 
                 del template["Resources"][logical_id]
                 for resource in translated:
-                    if verify_unique_logical_id(resource, sam_template["Resources"]):
+                    conflicting_logical_id = generated_logical_ids.record(resource, logical_id)
+                    if conflicting_logical_id:
+                        self.document_errors.append(
+                            DuplicateLogicalIdException(
+                                logical_id, resource.logical_id, resource.resource_type, conflicting_logical_id
+                            )
+                        )
+                    elif verify_unique_logical_id(resource, sam_template["Resources"]):
                         # For each generated resource, pass through existing metadata that may exist on the original SAM resource.
                         _r = resource.to_dict()
                         if (
@@ -219,7 +231,22 @@ def translate(  # noqa: PLR0912, PLR0915
                     template.get("Conditions", {}).update(new_conditions)
 
             if not deployment_preference_collection.can_skip_service_role():
-                template["Resources"].update(deployment_preference_collection.get_codedeploy_iam_role().to_dict())
+                codedeploy_role = deployment_preference_collection.get_codedeploy_iam_role()
+                # The CodeDeploy service role is not generated by any single SAM resource, so record it under a source
+                # that no SAM logical id can match (logical ids are alphanumeric). Otherwise a collision with the
+                # execution role of a function that also has a DeploymentPreference would not be detected.
+                conflicting_logical_id = generated_logical_ids.record(codedeploy_role, CODEDEPLOY_SERVICE_ROLE_SOURCE)
+                if conflicting_logical_id:
+                    self.document_errors.append(
+                        InvalidResourceException(
+                            conflicting_logical_id,
+                            f"It generates an IAM role with id [{codedeploy_role.logical_id}], which is reserved for"
+                            " the CodeDeploy service role generated for DeploymentPreference."
+                            " Please use a different id for this resource.",
+                        )
+                    )
+                else:
+                    template["Resources"].update(codedeploy_role.to_dict())
 
             for logical_id in deployment_preference_collection.enabled_logical_ids():
                 try:

samtranslator/translator/verify_logical_id.py

@@ -1,3 +1,4 @@
+from collections.abc import Callable
 from typing import Any
 
 from samtranslator.model import Resource
@@ -36,3 +37,42 @@ def verify_unique_logical_id(resource: Resource, existing_resources: dict[str, A
         resource.resource_type in do_not_verify
         and existing_resources[resource.logical_id]["Type"] in do_not_verify[resource.resource_type]
     )
+
+
+def is_iam_role(resource: Resource) -> bool:
+    """Default scope for GeneratedLogicalIdTracker: only IAM roles are checked.
+
+    SAM never intentionally generates the same IAM role from two different SAM resources, so any such collision is a
+    bug that would silently give one resource another resource's permissions. Other resource types (e.g. shared API
+    usage plans, custom domains, Route53 record set groups) ARE intentionally generated by multiple SAM resources
+    under the same logical id, so they are not checked yet.
+    """
+    return resource.resource_type == "AWS::IAM::Role"
+
+
+class GeneratedLogicalIdTracker:
+    """Detects when different SAM resources generate resources with the same logical id.
+
+    Generated resources are merged into the output template with ``dict.update()``, so without this check a later
+    resource silently overwrites an earlier one with the same logical id. ``verify_unique_logical_id`` only guards
+    against collisions with resources written in the input template; this class guards against collisions between
+    generated resources.
+
+    Which generated resources are checked is controlled by ``should_check``, so the scope can be widened (e.g. to all
+    resource types minus an allowlist of intentionally shared logical ids) without changing the tracker or its callers.
+    """
+
+    def __init__(self, should_check: Callable[[Resource], bool] = is_iam_role) -> None:
+        self._should_check = should_check
+        self._sources: dict[str, str] = {}
+
+    def record(self, resource: Resource, source_logical_id: str) -> str | None:
+        """Record that ``source_logical_id`` generated ``resource``.
+
+        :return: the logical id of a *different* SAM resource that already generated a resource with the same
+            logical id, or None if there is no conflict (or the resource is out of scope).
+        """
+        if resource.logical_id is None or not self._should_check(resource):
+            return None
+        existing_source = self._sources.setdefault(resource.logical_id, source_logical_id)
+        return existing_source if existing_source != source_logical_id else None

tests/translator/input/error_codedeploy_role_logical_id_collision.yaml

@@ -0,0 +1,19 @@
+# Function "CodeDeployService" generates role "CodeDeployServiceRole", which is the
+# same logical id as the CodeDeploy service role generated for DeploymentPreference.
+Resources:
+  CodeDeployService:
+    Type: AWS::Serverless::Function
+    Properties:
+      CodeUri: s3://sam-demo-bucket/hello.zip
+      Handler: index.handler
+      Runtime: python3.12
+
+  MyFunction:
+    Type: AWS::Serverless::Function
+    Properties:
+      CodeUri: s3://sam-demo-bucket/hello.zip
+      Handler: index.handler
+      Runtime: python3.12
+      AutoPublishAlias: live
+      DeploymentPreference:
+        Type: AllAtOnce

tests/translator/input/error_codedeploy_role_logical_id_self_collision.yaml

@@ -0,0 +1,12 @@
+# Function "CodeDeployService" generates execution role "CodeDeployServiceRole" and also has a
+# DeploymentPreference, so the CodeDeploy service role generated for it has the same logical id.
+Resources:
+  CodeDeployService:
+    Type: AWS::Serverless::Function
+    Properties:
+      CodeUri: s3://sam-demo-bucket/hello.zip
+      Handler: index.handler
+      Runtime: python3.12
+      AutoPublishAlias: live
+      DeploymentPreference:
+        Type: AllAtOnce

tests/translator/input/error_generated_role_logical_id_collision.yaml

@@ -0,0 +1,20 @@
+# Function "NetOperator" generates role "NetOperatorRole" and NetworkConnector "Net"
+# generates operator role "NetOperatorRole". The transform must fail instead of
+# silently giving the function the NetworkConnector operator role.
+Resources:
+  NetOperator:
+    Type: AWS::Serverless::Function
+    Properties:
+      CodeUri: s3://sam-demo-bucket/hello.zip
+      Handler: index.handler
+      Runtime: python3.12
+
+  Net:
+    Type: AWS::Serverless::NetworkConnector
+    Properties:
+      VpcConfig:
+        SubnetIds:
+        - subnet-12345678
+        SecurityGroupIds:
+        - sg-12345678
+        NetworkProtocol: IPv4

tests/translator/output/error_codedeploy_role_logical_id_collision.json

@@ -0,0 +1,15 @@
+{
+  "_autoGeneratedBreakdownErrorMessage": [
+    "Invalid Serverless Application Specification document. ",
+    "Number of errors found: 1. ",
+    "Resource with id [CodeDeployService] is invalid. ",
+    "It generates an IAM role with id [CodeDeployServiceRole], which is reserved for the CodeDeploy service role generated for DeploymentPreference. ",
+    "Please use a different id for this resource."
+  ],
+  "errorMessage": "Invalid Serverless Application Specification document. Number of errors found: 1. Resource with id [CodeDeployService] is invalid. It generates an IAM role with id [CodeDeployServiceRole], which is reserved for the CodeDeploy service role generated for DeploymentPreference. Please use a different id for this resource.",
+  "errors": [
+    {
+      "errorMessage": "Resource with id [CodeDeployService] is invalid. It generates an IAM role with id [CodeDeployServiceRole], which is reserved for the CodeDeploy service role generated for DeploymentPreference. Please use a different id for this resource."
+    }
+  ]
+}

tests/translator/output/error_codedeploy_role_logical_id_self_collision.json

@@ -0,0 +1,15 @@
+{
+  "_autoGeneratedBreakdownErrorMessage": [
+    "Invalid Serverless Application Specification document. ",
+    "Number of errors found: 1. ",
+    "Resource with id [CodeDeployService] is invalid. ",
+    "It generates an IAM role with id [CodeDeployServiceRole], which is reserved for the CodeDeploy service role generated for DeploymentPreference. ",
+    "Please use a different id for this resource."
+  ],
+  "errorMessage": "Invalid Serverless Application Specification document. Number of errors found: 1. Resource with id [CodeDeployService] is invalid. It generates an IAM role with id [CodeDeployServiceRole], which is reserved for the CodeDeploy service role generated for DeploymentPreference. Please use a different id for this resource.",
+  "errors": [
+    {
+      "errorMessage": "Resource with id [CodeDeployService] is invalid. It generates an IAM role with id [CodeDeployServiceRole], which is reserved for the CodeDeploy service role generated for DeploymentPreference. Please use a different id for this resource."
+    }
+  ]
+}

tests/translator/output/error_generated_role_logical_id_collision.json

@@ -0,0 +1,15 @@
+{
+  "_autoGeneratedBreakdownErrorMessage": [
+    "Invalid Serverless Application Specification document. ",
+    "Number of errors found: 1. ",
+    "Transforming resource with id [Net] attempts to create a new resource with id [NetOperatorRole] and type \"AWS::IAM::Role\". ",
+    "Resource with id [NetOperator] already generates a resource with that id. ",
+    "Please use a different id for one of these resources."
+  ],
+  "errorMessage": "Invalid Serverless Application Specification document. Number of errors found: 1. Transforming resource with id [Net] attempts to create a new resource with id [NetOperatorRole] and type \"AWS::IAM::Role\". Resource with id [NetOperator] already generates a resource with that id. Please use a different id for one of these resources.",
+  "errors": [
+    {
+      "errorMessage": "Transforming resource with id [Net] attempts to create a new resource with id [NetOperatorRole] and type \"AWS::IAM::Role\". Resource with id [NetOperator] already generates a resource with that id. Please use a different id for one of these resources."
+    }
+  ]
+}

tests/translator/test_verify_logical_id.py

@@ -0,0 +1,40 @@
+from unittest import TestCase
+
+from samtranslator.model.apigateway import ApiGatewayUsagePlan
+from samtranslator.model.iam import IAMRole
+from samtranslator.translator.verify_logical_id import GeneratedLogicalIdTracker
+
+
+class TestGeneratedLogicalIdTracker(TestCase):
+    def test_returns_conflicting_source_for_role_generated_by_different_resource(self):
+        tracker = GeneratedLogicalIdTracker()
+
+        self.assertIsNone(tracker.record(IAMRole("NetOperatorRole"), "NetOperator"))
+        self.assertEqual(tracker.record(IAMRole("NetOperatorRole"), "Net"), "NetOperator")
+
+    def test_allows_same_resource_to_generate_role_again(self):
+        tracker = GeneratedLogicalIdTracker()
+
+        self.assertIsNone(tracker.record(IAMRole("MyFunctionRole"), "MyFunction"))
+        self.assertIsNone(tracker.record(IAMRole("MyFunctionRole"), "MyFunction"))
+
+    def test_keeps_first_source_after_conflict(self):
+        tracker = GeneratedLogicalIdTracker()
+
+        tracker.record(IAMRole("NetOperatorRole"), "NetOperator")
+        tracker.record(IAMRole("NetOperatorRole"), "Net")
+
+        self.assertEqual(tracker.record(IAMRole("NetOperatorRole"), "Other"), "NetOperator")
+
+    def test_ignores_non_role_resources_by_default(self):
+        tracker = GeneratedLogicalIdTracker()
+
+        self.assertIsNone(tracker.record(ApiGatewayUsagePlan("ServerlessUsagePlan"), "ApiA"))
+        self.assertIsNone(tracker.record(ApiGatewayUsagePlan("ServerlessUsagePlan"), "ApiB"))
+
+    def test_scope_can_be_widened(self):
+        tracker = GeneratedLogicalIdTracker(should_check=lambda resource: True)
+
+        tracker.record(ApiGatewayUsagePlan("ServerlessUsagePlan"), "ApiA")
+
+        self.assertEqual(tracker.record(ApiGatewayUsagePlan("ServerlessUsagePlan"), "ApiB"), "ApiA")