diff --git a/samcli/commands/local/cli_common/user_exceptions.py b/samcli/commands/local/cli_common/user_exceptions.py index 3ba5c2df1d4..6008d77a4d7 100644 --- a/samcli/commands/local/cli_common/user_exceptions.py +++ b/samcli/commands/local/cli_common/user_exceptions.py @@ -47,6 +47,12 @@ class ResourceNotFound(UserException): """ +class AmbiguousResourceIdentifier(UserException): + """ + A bare resource logical ID (without a stack path) matched resources in more than one stack + """ + + class InvalidLayerVersionArn(UserException): """ The LayerVersion Arn given in the template is Invalid diff --git a/samcli/lib/providers/provider.py b/samcli/lib/providers/provider.py index a200465fc41..b987a86f074 100644 --- a/samcli/lib/providers/provider.py +++ b/samcli/lib/providers/provider.py @@ -11,9 +11,10 @@ from dataclasses import dataclass from enum import Enum from pathlib import Path -from typing import Any, Dict, Iterator, List, NamedTuple, Optional, Set, Union, cast +from typing import Any, Dict, Iterator, List, NamedTuple, Optional, Set, Tuple, Union, cast from samcli.commands.local.cli_common.user_exceptions import ( + AmbiguousResourceIdentifier, InvalidFunctionPropertyType, InvalidLayerVersionArn, UnsupportedIntrinsic, @@ -883,22 +884,52 @@ def get_resource_by_id( ------- Dict Resource dict + + Raises + ------ + AmbiguousResourceIdentifier + If a bare logical ID (no stack path given) matches resources in more than one *nested* + stack, with no match in the root stack to prefer. The root stack, if it has a match, always + takes priority (this is relied upon by existing behavior); but silently picking the first + nested-stack match in stack-list order when there is no root match - and more than one + nested stack collides - could operate on the wrong physical resource (e.g. + `sam sync --resource-id`), so that specific case is now a clear, actionable error instead. """ search_all_stacks = not identifier.stack_path and not explicit_nested + matches: List[Tuple[Stack, str, Dict[str, Any]]] = [] for stack in stacks: if stack.stack_path == identifier.stack_path or search_all_stacks: found_resource = None + found_resource_id = None for logical_id, resource in stack.resources.items(): resource_id = ResourceMetadataNormalizer.get_resource_id(resource, logical_id) if resource_id == identifier.resource_iac_id or ( not identifier.stack_path and logical_id == identifier.resource_iac_id ): found_resource = resource + found_resource_id = resource_id break if found_resource: - return cast(Dict[str, Any], found_resource) - return None + if not stack.stack_path: + # The root stack always takes priority over nested stacks. + return cast(Dict[str, Any], found_resource) + matches.append((stack, cast(str, found_resource_id), found_resource)) + if not search_all_stacks: + break + + if len(matches) > 1: + # Build the remediation paths from the *matched* (normalized) resource ID, not the raw + # identifier the user typed: for CDK/SamResourceId resources the two can differ, and a + # path built from the raw logical ID would never resolve on retry. + colliding_paths = [get_full_path(stack.stack_path, resource_id) for stack, resource_id, _ in matches] + raise AmbiguousResourceIdentifier( + f"Resource ID '{identifier.resource_iac_id}' is ambiguous: it matches resources in more than one " + f"nested stack ({', '.join(colliding_paths)}). Qualify it with the full stack path, " + f"e.g. '{colliding_paths[0]}'." + ) + + return cast(Dict[str, Any], matches[0][2]) if matches else None def get_resource_full_path_by_id(stacks: List[Stack], identifier: ResourceIdentifier) -> Optional[str]: @@ -915,7 +946,28 @@ def get_resource_full_path_by_id(stacks: List[Stack], identifier: ResourceIdenti ------- str return resource full path + + Raises + ------ + AmbiguousResourceIdentifier + If a bare logical ID (no stack path given) matches resources in more than one *nested* + stack, with no match in the root stack to prefer. Mirrors the precedence/ambiguity rule + enforced by get_resource_by_id, so the two entry points cannot disagree on what a bare ID + resolves to. + + Note this function is also used outside `sam sync --resource-id` (package_context, + guided_context, image_repository_validation, all mapping a user-supplied image function + ID to an image repository URI): an ambiguous match there means the wrong function could + silently receive the wrong image repository, which is the same "operates on the wrong + physical resource" hazard this function exists to prevent for sync, so raising here is + intentional rather than sync-specific. This is a deliberately *stricter* policy than + `SamFunctionProvider.get()` (used by e.g. `sam local invoke`), which instead warns and + picks the alphabetically-first match; that entry point resolves a *running* local + function for interactive use, where a warning is recoverable, whereas the callers of this + function feed the result into a deploy/package/validation decision without a further + confirmation step. """ + matches: List[str] = [] for stack in stacks: if identifier.stack_path and identifier.stack_path != stack.stack_path: continue @@ -924,8 +976,19 @@ def get_resource_full_path_by_id(stacks: List[Stack], identifier: ResourceIdenti if resource_id == identifier.resource_iac_id or ( not identifier.stack_path and logical_id == identifier.resource_iac_id ): - return get_full_path(stack.stack_path, resource_id) - return None + if not stack.stack_path: + # The root stack always takes priority over nested stacks. + return get_full_path(stack.stack_path, resource_id) + matches.append(get_full_path(stack.stack_path, resource_id)) + break + + if len(matches) > 1: + raise AmbiguousResourceIdentifier( + f"Resource ID '{identifier.resource_iac_id}' is ambiguous: it matches resources in more than one " + f"nested stack ({', '.join(matches)}). Qualify it with the full stack path, e.g. '{matches[0]}'." + ) + + return matches[0] if matches else None def get_resource_ids_by_type(stacks: List[Stack], resource_type: str) -> List[ResourceIdentifier]: @@ -1045,8 +1108,7 @@ def get_function_build_info( loadable = imageuri and check_path_valid_type(imageuri) and Path(imageuri).is_file() if not buildable and not loadable: LOG.debug( - "Skip Building %s function, as it is missing either Dockerfile or DockerContext " - "metadata properties.", + "Skip Building %s function, as it is missing either Dockerfile or DockerContext metadata properties.", full_path, ) return FunctionBuildInfo.NonBuildableImage diff --git a/tests/unit/commands/local/lib/test_provider.py b/tests/unit/commands/local/lib/test_provider.py index 0e48c1d9e91..73e39c48a53 100644 --- a/tests/unit/commands/local/lib/test_provider.py +++ b/tests/unit/commands/local/lib/test_provider.py @@ -21,6 +21,7 @@ FunctionBuildInfo, ) from samcli.commands.local.cli_common.user_exceptions import ( + AmbiguousResourceIdentifier, InvalidLayerVersionArn, UnsupportedIntrinsic, InvalidFunctionPropertyType, @@ -652,6 +653,94 @@ def test_get_resource_by_id_not_found( self.assertEqual(result, None) +class TestGetResourceByIDAmbiguousNestedStacks(TestCase): + """ + Regression tests: a bare (unqualified) resource ID that matches resources in more than one + *nested* stack, with no root-stack resource to prefer, used to silently return whichever + stack happened to come first in the input list - which could point sync/invoke/logs operations + at the wrong deployed physical resource. This must now raise AmbiguousResourceIdentifier instead. + """ + + def setUp(self) -> None: + super().setUp() + self.root_stack = MagicMock() + self.root_stack.stack_path = "" + self.root_stack.resources = {} + + self.nested_stack_a = MagicMock() + self.nested_stack_a.stack_path = "NestedStackA" + self.nested_stack_a.resources = {"Function1": {"Properties": "BodyA"}} + + self.nested_stack_b = MagicMock() + self.nested_stack_b.stack_path = "NestedStackB" + self.nested_stack_b.resources = {"Function1": {"Properties": "BodyB"}} + + def test_raises_when_two_sibling_nested_stacks_collide(self): + resource_identifier = MagicMock() + resource_identifier.stack_path = "" + resource_identifier.resource_iac_id = "Function1" + + with self.assertRaises(AmbiguousResourceIdentifier): + get_resource_by_id([self.root_stack, self.nested_stack_a, self.nested_stack_b], resource_identifier, False) + + def test_does_not_raise_when_only_one_nested_stack_matches(self): + resource_identifier = MagicMock() + resource_identifier.stack_path = "" + resource_identifier.resource_iac_id = "Function1" + + result = get_resource_by_id([self.root_stack, self.nested_stack_a], resource_identifier, False) + self.assertEqual(result, self.nested_stack_a.resources["Function1"]) + + def test_does_not_raise_when_explicitly_qualified_with_stack_path(self): + resource_identifier = MagicMock() + resource_identifier.stack_path = "NestedStackA" + resource_identifier.resource_iac_id = "Function1" + + result = get_resource_by_id( + [self.root_stack, self.nested_stack_a, self.nested_stack_b], resource_identifier, False + ) + self.assertEqual(result, self.nested_stack_a.resources["Function1"]) + + def test_ambiguity_error_suggests_a_path_that_actually_resolves_for_cdk_resources(self): + """Regression test: the collision message must be built from the *normalized* resource ID + (ResourceMetadataNormalizer.get_resource_id, e.g. from a CDK resource's SamResourceId + metadata), not the raw logical ID the user typed. Otherwise, for a bare logical ID that + only matches via the SamResourceId fallback, the suggested qualified path + ('NestedStackA/') can never resolve, since a stack-qualified lookup only + matches on the normalized ID -- sending the user to a dead end. + """ + nested_stack_a = MagicMock() + nested_stack_a.stack_path = "NestedStackA" + nested_stack_a.resources = { + "Function1": {"Properties": "BodyA", "Metadata": {"SamResourceId": "Function1-x"}}, + } + nested_stack_b = MagicMock() + nested_stack_b.stack_path = "NestedStackB" + nested_stack_b.resources = { + "Function1": {"Properties": "BodyB", "Metadata": {"SamResourceId": "Function1-x"}}, + } + + resource_identifier = MagicMock() + resource_identifier.stack_path = "" + resource_identifier.resource_iac_id = "Function1" + + with self.assertRaises(AmbiguousResourceIdentifier) as ctx: + get_resource_by_id([self.root_stack, nested_stack_a, nested_stack_b], resource_identifier, False) + + message = str(ctx.exception) + self.assertIn("NestedStackA/Function1-x", message) + self.assertIn("NestedStackB/Function1-x", message) + # The un-resolvable, raw-logical-ID-qualified form must not appear as the suggestion. + self.assertNotIn("NestedStackA/Function1'", message) + + # And the suggested qualified path must actually resolve. + qualified_identifier = MagicMock() + qualified_identifier.stack_path = "NestedStackA" + qualified_identifier.resource_iac_id = "Function1-x" + result = get_resource_by_id([self.root_stack, nested_stack_a, nested_stack_b], qualified_identifier, False) + self.assertEqual(result, nested_stack_a.resources["Function1"]) + + class TestGetResourceIDsByType(TestCase): def setUp(self) -> None: super().setUp() @@ -863,6 +952,67 @@ def test_get_resource_full_path_by_id(self, resource_id, expected_full_path): self.assertEqual(expected_full_path, full_path) +class TestGetResourceFullPathByIDAmbiguousNestedStacks(TestCase): + """ + Regression tests mirroring TestGetResourceByIDAmbiguousNestedStacks: get_resource_full_path_by_id + must apply the same root-stack-priority and nested-stack-ambiguity rules as get_resource_by_id, + so the two entry points cannot disagree on what a bare resource ID resolves to. + """ + + def setUp(self) -> None: + super().setUp() + self.root_stack = MagicMock() + self.root_stack.stack_path = "" + self.root_stack.resources = {} + + self.nested_stack_a = MagicMock() + self.nested_stack_a.stack_path = "NestedStackA" + self.nested_stack_a.resources = {"Function1": {"Properties": "BodyA"}} + + self.nested_stack_b = MagicMock() + self.nested_stack_b.stack_path = "NestedStackB" + self.nested_stack_b.resources = {"Function1": {"Properties": "BodyB"}} + + def test_raises_when_two_sibling_nested_stacks_collide(self): + resource_identifier = MagicMock() + resource_identifier.stack_path = "" + resource_identifier.resource_iac_id = "Function1" + + with self.assertRaises(AmbiguousResourceIdentifier): + get_resource_full_path_by_id( + [self.root_stack, self.nested_stack_a, self.nested_stack_b], resource_identifier + ) + + def test_does_not_raise_when_only_one_nested_stack_matches(self): + resource_identifier = MagicMock() + resource_identifier.stack_path = "" + resource_identifier.resource_iac_id = "Function1" + + result = get_resource_full_path_by_id([self.root_stack, self.nested_stack_a], resource_identifier) + self.assertEqual(result, "NestedStackA/Function1") + + def test_does_not_raise_when_explicitly_qualified_with_stack_path(self): + resource_identifier = MagicMock() + resource_identifier.stack_path = "NestedStackA" + resource_identifier.resource_iac_id = "Function1" + + result = get_resource_full_path_by_id( + [self.root_stack, self.nested_stack_a, self.nested_stack_b], resource_identifier + ) + self.assertEqual(result, "NestedStackA/Function1") + + def test_root_stack_takes_priority_over_nested_match(self): + self.root_stack.resources = {"Function1": {"Properties": "RootBody"}} + resource_identifier = MagicMock() + resource_identifier.stack_path = "" + resource_identifier.resource_iac_id = "Function1" + + result = get_resource_full_path_by_id( + [self.nested_stack_a, self.nested_stack_b, self.root_stack], resource_identifier + ) + self.assertEqual(result, "Function1") + + class TestGetStack(TestCase): root_stack = Stack("", "Root", "template.yaml", None, {}) child_stack = Stack("Root", "Child", "root_stack/template.yaml", None, {})