Skip to content
Open
Show file tree
Hide file tree
Changes from 3 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions samcli/commands/local/cli_common/user_exceptions.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
64 changes: 57 additions & 7 deletions samcli/lib/providers/provider.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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(

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[GENERAL] The same "bare ID matches resources in more than one stack" condition is already handled elsewhere in the codebase with the opposite strategy — warn and pick deterministically (samcli/lib/providers/sam_function_provider.py:139-160):

# If multiple functions are found, only return one of them
if len(found_fs) > 1:
   found_fs.sort(key=lambda f0: f0.full_path.lower())
   message = (
       f"Multiple functions found with keyword {name}! Function {found_fs[0].full_path} will be "
       f"invoked! If it's not the function you are going to invoke, please choose one of them from below:"
   )
   LOG.warning(Colored().yellow(message))
   resolved_function = found_fs[0]

After this change the two resolvers disagree on the same user input: sam sync --resource-id Function1 hard-fails, while sam local invoke Function1 still silently picks the alphabetically-first full path with a warning. That matters for two reasons: the "operates on the wrong resource" hazard motivating this PR is only closed on one of the two resolution paths, and future maintainers now have two contradictory precedence rules to reason about (first-nested-match is an error here; lowest-sorted-full-path wins there).

Either align the two (reuse the warn-and-pick behavior, or extend the ambiguity error to SamFunctionProvider.get), or add a short comment next to this raise explaining why sync-style resolution must be stricter than invoke-style resolution, so the divergence is a documented decision rather than an accident.

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]:
Expand All @@ -915,7 +946,16 @@ 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.
"""
matches: List[str] = []
for stack in stacks:
if identifier.stack_path and identifier.stack_path != stack.stack_path:
continue
Expand All @@ -924,8 +964,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(

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[GENERAL] Raising from get_resource_full_path_by_id extends this change's blast radius well past sam sync, and turns some previously-working invocations into hard failures. The three callers of this function all resolve a user-supplied bare ID and are written to tolerate an unresolved key:

# samcli/commands/package/package_context.py:140
# samcli/commands/deploy/guided_context.py:358
repo_full_path = get_resource_full_path_by_id(stacks, ResourceIdentifier(image_repo_func_id))
if repo_full_path:
   updated_repo[repo_full_path] = image_repo_uri

samcli/lib/cli_validation/image_repository_validation.py:138 is similar — it collects results into a set and compares against the image-function full paths, expecting a clean click.BadOptionUsage when the sets differ.

So for a template with two sibling nested stacks each containing Function1, sam deploy --image-repositories Function1=<uri> and sam package previously resolved to the first match and continued; they now abort with AmbiguousResourceIdentifier. For sam deploy --guided the abort lands mid-flow, after the interactive prompts have already been answered.

Erroring here is defensible (mapping an image repo to the wrong function is the same class of hazard as syncing the wrong function), but it is a user-facing behavior change for deploy/package, not just for the sync --resource-id path the PR describes. Worth either scoping the hard error to the sync resolution path or calling the deploy/package impact out explicitly in the PR description/changelog so it isn't discovered as a regression.

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]:
Expand Down Expand Up @@ -1045,8 +1096,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
Expand Down
150 changes: 150 additions & 0 deletions tests/unit/commands/local/lib/test_provider.py
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@
FunctionBuildInfo,
)
from samcli.commands.local.cli_common.user_exceptions import (
AmbiguousResourceIdentifier,
InvalidLayerVersionArn,
UnsupportedIntrinsic,
InvalidFunctionPropertyType,
Expand Down Expand Up @@ -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/<logicalId>') 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()
Expand Down Expand Up @@ -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, {})
Expand Down