diff --git a/samtranslator/model/exceptions.py b/samtranslator/model/exceptions.py index ad0603f38..f1ac8291c 100644 --- a/samtranslator/model/exceptions.py +++ b/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' diff --git a/samtranslator/translator/translator.py b/samtranslator/translator/translator.py index 2aeadbe2c..eabf7b304 100644 --- a/samtranslator/translator/translator.py +++ b/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: diff --git a/samtranslator/translator/verify_logical_id.py b/samtranslator/translator/verify_logical_id.py index 7da2c7911..1f78edbb5 100644 --- a/samtranslator/translator/verify_logical_id.py +++ b/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 diff --git a/tests/translator/input/error_codedeploy_role_logical_id_collision.yaml b/tests/translator/input/error_codedeploy_role_logical_id_collision.yaml new file mode 100644 index 000000000..d95d3046b --- /dev/null +++ b/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 diff --git a/tests/translator/input/error_codedeploy_role_logical_id_self_collision.yaml b/tests/translator/input/error_codedeploy_role_logical_id_self_collision.yaml new file mode 100644 index 000000000..aaf95c492 --- /dev/null +++ b/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 diff --git a/tests/translator/input/error_generated_role_logical_id_collision.yaml b/tests/translator/input/error_generated_role_logical_id_collision.yaml new file mode 100644 index 000000000..a02ebf843 --- /dev/null +++ b/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 diff --git a/tests/translator/output/error_codedeploy_role_logical_id_collision.json b/tests/translator/output/error_codedeploy_role_logical_id_collision.json new file mode 100644 index 000000000..fb3522691 --- /dev/null +++ b/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." + } + ] +} diff --git a/tests/translator/output/error_codedeploy_role_logical_id_self_collision.json b/tests/translator/output/error_codedeploy_role_logical_id_self_collision.json new file mode 100644 index 000000000..fb3522691 --- /dev/null +++ b/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." + } + ] +} diff --git a/tests/translator/output/error_generated_role_logical_id_collision.json b/tests/translator/output/error_generated_role_logical_id_collision.json new file mode 100644 index 000000000..37da5ff66 --- /dev/null +++ b/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." + } + ] +} diff --git a/tests/translator/test_verify_logical_id.py b/tests/translator/test_verify_logical_id.py new file mode 100644 index 000000000..c938a2ea3 --- /dev/null +++ b/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")