From 64dfdf008143947ce3302b63b265d972b081e26b Mon Sep 17 00:00:00 2001 From: Sanjay Santhanam Date: Wed, 9 Sep 2026 11:30:16 -0700 Subject: [PATCH 1/6] fix: use service role for companion stack create/update The companion stack created by sam deploy --resolve-image-repos ignored the --role-arn service role: create_stack/update_stack were called without RoleARN, so ECR repo creation fell back to the caller's permissions and could fail with ecr:CreateRepository AccessDenied. Thread role_arn through CompanionStackManager and sync_ecr_stack, and pass RoleARN to CloudFormation create_stack/update_stack when set. Fixes #5051 --- samcli/commands/deploy/command.py | 2 +- .../companion_stack_manager.py | 36 ++++++++--- .../test_companion_stack_manager.py | 61 ++++++++++++++++++- 3 files changed, 87 insertions(+), 12 deletions(-) diff --git a/samcli/commands/deploy/command.py b/samcli/commands/deploy/command.py index 80f9c7e80e3..175eb8c62b4 100644 --- a/samcli/commands/deploy/command.py +++ b/samcli/commands/deploy/command.py @@ -362,7 +362,7 @@ def do_cli( # after we figure out how to enable resolve-images-repos in package if resolve_image_repos: image_repositories = sync_ecr_stack( - template_file, stack_name, region, s3_bucket, s3_prefix, image_repositories + template_file, stack_name, region, s3_bucket, s3_prefix, image_repositories, role_arn=role_arn ) with osutils.tempfile_platform_independent() as output_template_file: if guided: diff --git a/samcli/lib/bootstrap/companion_stack/companion_stack_manager.py b/samcli/lib/bootstrap/companion_stack/companion_stack_manager.py index 66dd6a290e1..5ca605c8ff8 100644 --- a/samcli/lib/bootstrap/companion_stack/companion_stack_manager.py +++ b/samcli/lib/bootstrap/companion_stack/companion_stack_manager.py @@ -3,7 +3,7 @@ """ import logging -from typing import Dict, List, Optional +from typing import Any, Dict, List, Optional import boto3 from botocore.config import Config @@ -41,7 +41,7 @@ class CompanionStackManager: _cfn_client: CloudFormationClient _s3_client: S3Client - def __init__(self, stack_name, region, s3_bucket, s3_prefix): + def __init__(self, stack_name, region, s3_bucket, s3_prefix, role_arn: Optional[str] = None): self._companion_stack = CompanionStack(stack_name) self._builder = CompanionStackBuilder(self._companion_stack) self._boto_config = Config(region_name=region if region else None) @@ -49,6 +49,7 @@ def __init__(self, stack_name, region, s3_bucket, s3_prefix): self._delete_stack_waiter_config = {"Delay": 10, "MaxAttempts": 120} self._s3_bucket = s3_bucket self._s3_prefix = s3_prefix + self._role_arn = role_arn try: self._cfn_client = boto3.client("cloudformation", config=self._boto_config) self._ecr_client = boto3.client("ecr", config=self._boto_config) @@ -117,16 +118,21 @@ def update_companion_stack(self) -> None: template_url = s3_uploader.to_path_style_s3_url(parts["Key"], parts.get("Version", None)) exists = self.does_companion_stack_exist() + # Use the service role passed via --role-arn (if any) so the companion stack is created/updated + # with the same permissions as the main stack. + stack_kwargs: Dict[str, Any] = { + "StackName": stack_name, + "TemplateURL": template_url, + "Capabilities": ["CAPABILITY_AUTO_EXPAND"], + } + if self._role_arn: + stack_kwargs["RoleARN"] = self._role_arn if exists: - self._cfn_client.update_stack( - StackName=stack_name, TemplateURL=template_url, Capabilities=["CAPABILITY_AUTO_EXPAND"] - ) + self._cfn_client.update_stack(**stack_kwargs) update_waiter = self._cfn_client.get_waiter("stack_update_complete") update_waiter.wait(StackName=stack_name, WaiterConfig=self._update_stack_waiter_config) else: - self._cfn_client.create_stack( - StackName=stack_name, TemplateURL=template_url, Capabilities=["CAPABILITY_AUTO_EXPAND"] - ) + self._cfn_client.create_stack(**stack_kwargs) create_waiter = self._cfn_client.get_waiter("stack_create_complete") create_waiter.wait(StackName=stack_name, WaiterConfig=self._update_stack_waiter_config) @@ -279,7 +285,13 @@ def is_repo_uri(self, repo_uri: Optional[str], function_logical_id: str) -> bool def sync_ecr_stack( - template_file: str, stack_name: str, region: str, s3_bucket: str, s3_prefix: str, image_repositories: Dict[str, str] + template_file: str, + stack_name: str, + region: str, + s3_bucket: str, + s3_prefix: str, + image_repositories: Dict[str, str], + role_arn: Optional[str] = None, ) -> Dict[str, str]: """Blocking call to sync local functions with ECR Companion Stack @@ -297,6 +309,10 @@ def sync_ecr_stack( S3 prefix for the bucket image_repositories : Dict[str, str] Mapping between function logical ID and ECR URI + role_arn : Optional[str] + Optional service role ARN used when creating or updating the companion stack. + When provided, the companion stack is created/updated with the same service + role as the main stack. Returns ------- @@ -305,7 +321,7 @@ def sync_ecr_stack( for Functions without a repo specified. """ image_repositories = image_repositories.copy() if image_repositories else {} - manager = CompanionStackManager(stack_name, region, s3_bucket, s3_prefix) + manager = CompanionStackManager(stack_name, region, s3_bucket, s3_prefix, role_arn=role_arn) stacks = SamLocalStackProvider.get_stacks(template_file, language_extensions_enabled=False)[0] function_provider = SamFunctionProvider(stacks, ignore_code_extraction_warnings=True) diff --git a/tests/unit/lib/bootstrap/companion_stack/test_companion_stack_manager.py b/tests/unit/lib/bootstrap/companion_stack/test_companion_stack_manager.py index 69f7d76ff65..c6c07036834 100644 --- a/tests/unit/lib/bootstrap/companion_stack/test_companion_stack_manager.py +++ b/tests/unit/lib/bootstrap/companion_stack/test_companion_stack_manager.py @@ -94,6 +94,45 @@ def test_update_companion_stack( self.cfn_client.get_waiter.assert_called_once_with("stack_update_complete") cfn_waiter.wait.assert_called_once_with(StackName=self.companion_stack_name, WaiterConfig=ANY) + def test_create_companion_stack_with_role_arn( + self, + ): + self._test_companion_stack_with_role_arn(exists=False) + + def test_update_companion_stack_with_role_arn( + self, + ): + self._test_companion_stack_with_role_arn(exists=True) + + @patch("samcli.lib.bootstrap.companion_stack.companion_stack_manager.mktempfile") + @patch("samcli.lib.bootstrap.companion_stack.companion_stack_manager.S3Uploader") + @patch("samcli.lib.bootstrap.companion_stack.companion_stack_manager.parse_s3_url") + def _test_companion_stack_with_role_arn( + self, + parse_s3_url_mock, + s3_uploader_mock, + mktempfile_mock, + exists, + ): + role_arn = "arn:aws:iam::123456789012:role/CloudFormationServiceRole" + self.boto3_client_mock.side_effect = [self.cfn_client, self.ecr_client, self.s3_client, self.sts_client] + manager = CompanionStackManager(self.stack_name, "region", "s3_bucket", "s3_prefix", role_arn=role_arn) + + cfn_waiter = Mock() + self.cfn_client.get_waiter.return_value = cfn_waiter + + manager.does_companion_stack_exist = lambda: exists + manager.update_companion_stack() + + stack_call = self.cfn_client.update_stack if exists else self.cfn_client.create_stack + stack_call.assert_called_once_with( + StackName=self.companion_stack_name, TemplateURL=ANY, Capabilities=ANY, RoleARN=role_arn + ) + self.cfn_client.get_waiter.assert_called_once_with( + "stack_update_complete" if exists else "stack_create_complete" + ) + cfn_waiter.wait.assert_called_once_with(StackName=self.companion_stack_name, WaiterConfig=ANY) + def test_delete_companion_stack(self): cfn_waiter = Mock() self.cfn_client.get_waiter.return_value = cfn_waiter @@ -276,7 +315,27 @@ def test_sync_ecr_stack(self, function_provider_mock, stack_provider_mock, manag result = sync_ecr_stack("template.yaml", "stack-name", "region", "s3-bucket", "s3-prefix", image_repositories) - manager_mock.assert_called_once_with("stack-name", "region", "s3-bucket", "s3-prefix") + manager_mock.assert_called_once_with("stack-name", "region", "s3-bucket", "s3-prefix", role_arn=None) + function_provider_mock.assert_called_once_with(stacks, ignore_code_extraction_warnings=True) + manager_mock.return_value.sync_repos.assert_called_once_with() + + self.assertEqual(result, {"Function1": "uri1", "Function2": "uri2"}) + + @patch("samcli.lib.bootstrap.companion_stack.companion_stack_manager.CompanionStackManager") + @patch("samcli.lib.bootstrap.companion_stack.companion_stack_manager.SamLocalStackProvider") + @patch("samcli.lib.bootstrap.companion_stack.companion_stack_manager.SamFunctionProvider") + def test_sync_ecr_stack_with_role_arn(self, function_provider_mock, stack_provider_mock, manager_mock): + image_repositories = {"Function1": "uri1"} + stacks = MagicMock() + stack_provider_mock.get_stacks.return_value = (stacks, None) + manager_mock.return_value.get_repository_mapping.return_value = {"Function2": "uri2"} + role_arn = "arn:aws:iam::123456789012:role/CloudFormationServiceRole" + + result = sync_ecr_stack( + "template.yaml", "stack-name", "region", "s3-bucket", "s3-prefix", image_repositories, role_arn=role_arn + ) + + manager_mock.assert_called_once_with("stack-name", "region", "s3-bucket", "s3-prefix", role_arn=role_arn) function_provider_mock.assert_called_once_with(stacks, ignore_code_extraction_warnings=True) manager_mock.return_value.sync_repos.assert_called_once_with() From 4bd8f4103ae096ea702a5d6d0b29acfe0f974220 Mon Sep 17 00:00:00 2001 From: Sanjay Santhanam Date: Wed, 9 Sep 2026 11:46:35 -0700 Subject: [PATCH 2/6] Address review: forward role_arn in guided deploy flow GuidedContext now accepts role_arn and forwards it at both guided companion-stack call sites (sync_ecr_stack in guided_prompts and CompanionStackManager in prompt_image_repository); do_cli passes --role-arn through on the guided path. Previously sam deploy --guided --role-arn --resolve-image-repos created the companion stack with caller credentials. Add guided role_arn tests. --- samcli/commands/deploy/command.py | 1 + samcli/commands/deploy/guided_context.py | 14 ++- .../commands/deploy/test_guided_context.py | 117 ++++++++++++++++++ 3 files changed, 130 insertions(+), 2 deletions(-) diff --git a/samcli/commands/deploy/command.py b/samcli/commands/deploy/command.py index 175eb8c62b4..66dfc586af2 100644 --- a/samcli/commands/deploy/command.py +++ b/samcli/commands/deploy/command.py @@ -341,6 +341,7 @@ def do_cli( config_file=config_file, disable_rollback=disable_rollback, language_extensions_enabled=language_extensions_enabled, + role_arn=role_arn, ) guided_context.run() else: diff --git a/samcli/commands/deploy/guided_context.py b/samcli/commands/deploy/guided_context.py index 1a3335687ae..e2d8ded0faf 100644 --- a/samcli/commands/deploy/guided_context.py +++ b/samcli/commands/deploy/guided_context.py @@ -63,6 +63,7 @@ def __init__( config_file=None, disable_rollback=None, language_extensions_enabled: bool = False, + role_arn: Optional[str] = None, ): self.template_file = template_file self.stack_name = stack_name @@ -97,6 +98,9 @@ def __init__( self.function_provider: Optional[SamFunctionProvider] = None self.disable_rollback = disable_rollback self._language_extensions_enabled = language_extensions_enabled + # Passed through to companion-stack operations so guided deploys honor + # --role-arn, matching the non-guided path. + self.role_arn = role_arn @property def guided_capabilities(self): @@ -189,7 +193,13 @@ def guided_prompts(self, parameter_override_keys): image_repositories = ( sync_ecr_stack( - self.template_file, stack_name, region, managed_s3_bucket, self.s3_prefix, self.image_repositories + self.template_file, + stack_name, + region, + managed_s3_bucket, + self.s3_prefix, + self.image_repositories, + role_arn=self.role_arn, ) if self.resolve_image_repositories else self.prompt_image_repository( @@ -359,7 +369,7 @@ def prompt_image_repository( if repo_full_path: updated_repositories[repo_full_path] = image_repo_uri self.function_provider = SamFunctionProvider(stacks, ignore_code_extraction_warnings=True) - manager = CompanionStackManager(stack_name, region, s3_bucket, s3_prefix) + manager = CompanionStackManager(stack_name, region, s3_bucket, s3_prefix, role_arn=self.role_arn) function_logical_ids = [ function.full_path for function in self.function_provider.get_all() if function.packagetype == IMAGE diff --git a/tests/unit/commands/deploy/test_guided_context.py b/tests/unit/commands/deploy/test_guided_context.py index 052d8a84a76..d0e3682668f 100644 --- a/tests/unit/commands/deploy/test_guided_context.py +++ b/tests/unit/commands/deploy/test_guided_context.py @@ -1002,3 +1002,120 @@ def test_guided_prompts_check_default_config_region( ), ] self.assertEqual(expected_prompt_calls, patched_prompt.call_args_list) + + +class TestGuidedContextRoleArn(TestCase): + ROLE_ARN = "arn:aws:iam::123456789012:role/deploy-role" + + def setUp(self): + self.gc = GuidedContext( + template_file="template", + stack_name="test", + s3_bucket="s3_b", + s3_prefix="s3_p", + confirm_changeset=True, + region="region", + image_repository=None, + image_repositories={"RandomFunction": "image-repo"}, + disable_rollback=False, + role_arn=self.ROLE_ARN, + ) + self.companion_stack_manager_patch = patch("samcli.commands.deploy.guided_context.CompanionStackManager") + self.companion_stack_manager_mock = self.companion_stack_manager_patch.start() + self.sync_ecr_stack_patch = patch("samcli.commands.deploy.guided_context.sync_ecr_stack") + self.sync_ecr_stack_mock = self.sync_ecr_stack_patch.start() + self.addCleanup(self.companion_stack_manager_patch.stop) + self.addCleanup(self.sync_ecr_stack_patch.stop) + + def test_guided_context_stores_role_arn(self): + self.assertEqual(self.gc.role_arn, self.ROLE_ARN) + default_gc = GuidedContext( + template_file="template", + stack_name="test", + s3_bucket="s3_b", + s3_prefix="s3_p", + region="region", + image_repository=None, + image_repositories={}, + ) + self.assertIsNone(default_gc.role_arn) + + @patch("samcli.commands.deploy.guided_context.get_resource_full_path_by_id") + @patch("samcli.commands.deploy.guided_context.prompt") + @patch("samcli.commands.deploy.guided_context.confirm") + @patch("samcli.commands.deploy.guided_context.manage_stack") + @patch("samcli.commands.deploy.guided_context.auth_per_resource") + @patch("samcli.commands.deploy.guided_context.SamLocalStackProvider.get_stacks") + @patch("samcli.commands.deploy.guided_context.signer_config_per_function") + def test_guided_prompts_forwards_role_arn_to_sync_ecr_stack( + self, + patched_signer_config_per_function, + patched_get_buildable_stacks, + patched_auth_per_resource, + patched_manage_stack, + patched_confirm, + patched_prompt, + get_resource_full_path_by_id_mock, + ): + patched_get_buildable_stacks.return_value = (Mock(), []) + patched_signer_config_per_function.return_value = (None, None) + patched_prompt.side_effect = ["sam-app", "region", "CAPABILITY_IAM", "samconfig.toml", "default"] + patched_confirm.side_effect = [True, True, False, True] + patched_auth_per_resource.return_value = [("HelloWorldFunction", True)] + patched_manage_stack.return_value = "managed_s3_stack" + self.sync_ecr_stack_mock.return_value = {"HelloWorldFunction": "repo-uri"} + + self.gc.resolve_image_repositories = True + self.gc.guided_prompts(parameter_override_keys=None) + + self.sync_ecr_stack_mock.assert_called_once_with( + "template", + "sam-app", + "region", + "managed_s3_stack", + "s3_p", + {"RandomFunction": "image-repo"}, + role_arn=self.ROLE_ARN, + ) + + @patch("samcli.commands.deploy.guided_context.get_resource_full_path_by_id") + @patch("samcli.commands.deploy.guided_context.prompt") + @patch("samcli.commands.deploy.guided_context.confirm") + @patch("samcli.commands.deploy.guided_context.manage_stack") + @patch("samcli.commands.deploy.guided_context.auth_per_resource") + @patch("samcli.commands.deploy.guided_context.SamLocalStackProvider.get_stacks") + @patch("samcli.commands.deploy.guided_context.SamFunctionProvider") + @patch("samcli.commands.deploy.guided_context.signer_config_per_function") + def test_prompt_image_repository_forwards_role_arn_to_companion_stack_manager( + self, + patched_signer_config_per_function, + patched_sam_function_provider, + patched_get_buildable_stacks, + patched_auth_per_resource, + patched_manage_stack, + patched_confirm, + patched_prompt, + get_resource_full_path_by_id_mock, + ): + function_mock = MagicMock() + function_mock.packagetype = IMAGE + function_mock.imageuri = "helloworld:v1" + function_mock.full_path = "HelloWorldFunction" + patched_sam_function_provider.return_value.get_all.return_value = [function_mock] + patched_get_buildable_stacks.return_value = (Mock(), []) + patched_signer_config_per_function.return_value = (None, None) + patched_prompt.side_effect = ["sam-app", "region", "CAPABILITY_IAM", "samconfig.toml", "default"] + patched_confirm.side_effect = [True, True, False, True, True] + patched_auth_per_resource.return_value = [("HelloWorldFunction", True)] + get_resource_full_path_by_id_mock.return_value = None + patched_manage_stack.return_value = "managed_s3_stack" + # "do not create repos for all functions" then "do not delete unreferenced repos" + manager_instance = self.companion_stack_manager_mock.return_value + manager_instance.get_repository_mapping.return_value = {"HelloWorldFunction": "repo-uri"} + manager_instance.get_unreferenced_repos.return_value = [] + + self.gc.guided_prompts(parameter_override_keys=None) + + self.companion_stack_manager_mock.assert_called_once_with( + "sam-app", "region", "managed_s3_stack", "s3_p", role_arn=self.ROLE_ARN + ) From 5fdfcc738905e4a9c41d53efdbbd62deb06ba449 Mon Sep 17 00:00:00 2001 From: Sanjay Santhanam Date: Wed, 9 Sep 2026 11:54:55 -0700 Subject: [PATCH 3/6] Address bot review: assume role_arn for direct ECR calls in companion stack manager --- .../companion_stack_manager.py | 48 ++++++++++++- .../test_companion_stack_manager.py | 68 ++++++++++++++++++- 2 files changed, 113 insertions(+), 3 deletions(-) diff --git a/samcli/lib/bootstrap/companion_stack/companion_stack_manager.py b/samcli/lib/bootstrap/companion_stack/companion_stack_manager.py index 5ca605c8ff8..71c8cdc7517 100644 --- a/samcli/lib/bootstrap/companion_stack/companion_stack_manager.py +++ b/samcli/lib/bootstrap/companion_stack/companion_stack_manager.py @@ -54,7 +54,14 @@ def __init__(self, stack_name, region, s3_bucket, s3_prefix, role_arn: Optional[ self._cfn_client = boto3.client("cloudformation", config=self._boto_config) self._ecr_client = boto3.client("ecr", config=self._boto_config) self._s3_client = boto3.client("s3", config=self._boto_config) - self._account_id = boto3.client("sts").get_caller_identity().get("Account") + sts_client = boto3.client("sts") + self._account_id = sts_client.get_caller_identity().get("Account") + if role_arn: + # The CloudFormation RoleARN only covers calls CloudFormation + # makes on our behalf, not SDK calls SAM CLI makes itself + # (e.g. deleting unreferenced ECR repos). Assume the same role + # so all companion-stack side effects use one identity. + self._ecr_client = self._ecr_client_for_role(sts_client, role_arn) self._region_name = self._cfn_client.meta.region_name except NoCredentialsError as ex: raise AWSServiceClientError( @@ -69,6 +76,42 @@ def __init__(self, stack_name, region, s3_bucket, s3_prefix, role_arn: Optional[ "Please provide a region via the --region parameter or by the AWS_DEFAULT_REGION environment variable." ) from ex + def _ecr_client_for_role(self, sts_client, role_arn: str): + """ + Assume the deployment service role and return an ECR client that uses + the assumed-role credentials. + + Parameters + ---------- + sts_client + STS client used to assume the role + role_arn : str + ARN of the service role (e.g. from --role-arn) to assume + + Returns + ------- + An ECR client authenticated with the assumed-role credentials + + Raises + ------ + AWSServiceClientError + If the role cannot be assumed + """ + try: + assumed_role = sts_client.assume_role(RoleArn=role_arn, RoleSessionName="sam-cli-companion-stack") + except ClientError as ex: + raise AWSServiceClientError( + f"Error assuming the provided role {role_arn} for companion stack ECR operations: {ex}" + ) from ex + credentials = assumed_role["Credentials"] + return boto3.client( + "ecr", + config=self._boto_config, + aws_access_key_id=credentials["AccessKeyId"], + aws_secret_access_key=credentials["SecretAccessKey"], + aws_session_token=credentials["SessionToken"], + ) + def set_functions( self, function_logical_ids: List[str], image_repositories: Optional[Dict[str, str]] = None ) -> None: @@ -312,7 +355,8 @@ def sync_ecr_stack( role_arn : Optional[str] Optional service role ARN used when creating or updating the companion stack. When provided, the companion stack is created/updated with the same service - role as the main stack. + role as the main stack, and direct ECR calls made by SAM CLI (e.g. deleting + unreferenced repositories) assume the role so they run under one identity. Returns ------- diff --git a/tests/unit/lib/bootstrap/companion_stack/test_companion_stack_manager.py b/tests/unit/lib/bootstrap/companion_stack/test_companion_stack_manager.py index c6c07036834..d2acb98d707 100644 --- a/tests/unit/lib/bootstrap/companion_stack/test_companion_stack_manager.py +++ b/tests/unit/lib/bootstrap/companion_stack/test_companion_stack_manager.py @@ -115,9 +115,27 @@ def _test_companion_stack_with_role_arn( exists, ): role_arn = "arn:aws:iam::123456789012:role/CloudFormationServiceRole" - self.boto3_client_mock.side_effect = [self.cfn_client, self.ecr_client, self.s3_client, self.sts_client] + role_ecr_client = Mock() + self.sts_client.assume_role.return_value = { + "Credentials": { + "AccessKeyId": "AKIAIOSFODNN7EXAMPLE", + "SecretAccessKey": "secret", + "SessionToken": "token", + } + } + self.boto3_client_mock.side_effect = [ + self.cfn_client, + self.ecr_client, + self.s3_client, + self.sts_client, + role_ecr_client, + ] manager = CompanionStackManager(self.stack_name, "region", "s3_bucket", "s3_prefix", role_arn=role_arn) + # the service role is assumed and the ECR client uses its credentials + self.sts_client.assume_role.assert_called_once_with(RoleArn=role_arn, RoleSessionName="sam-cli-companion-stack") + self.assertIs(manager._ecr_client, role_ecr_client) + cfn_waiter = Mock() self.cfn_client.get_waiter.return_value = cfn_waiter @@ -251,6 +269,54 @@ def test_delete_unreferenced_repos(self): self.ecr_client.delete_repository.assert_any_call(repositoryName=repo_a_id, force=True) self.ecr_client.delete_repository.assert_any_call(repositoryName=repo_b_id, force=True) + def test_delete_unreferenced_repos_with_role_arn_uses_assumed_role_client(self): + # With --role-arn set, direct ECR calls (delete_unreferenced_repos) + # must go through the assumed-role client, not the caller's. + role_arn = "arn:aws:iam::123456789012:role/CloudFormationServiceRole" + role_ecr_client = Mock() + self.sts_client.assume_role.return_value = { + "Credentials": { + "AccessKeyId": "AKIAIOSFODNN7EXAMPLE", + "SecretAccessKey": "secret", + "SessionToken": "token", + } + } + self.boto3_client_mock.side_effect = [ + self.cfn_client, + self.ecr_client, + self.s3_client, + self.sts_client, + role_ecr_client, + ] + manager = CompanionStackManager(self.stack_name, "region", "s3_bucket", "s3_prefix", role_arn=role_arn) + + repo = Mock() + repo.physical_id = "ECRRepoStale" + manager.get_unreferenced_repos = lambda: [repo] + + manager.delete_unreferenced_repos() + + role_ecr_client.delete_repository.assert_called_once_with(repositoryName="ECRRepoStale", force=True) + self.ecr_client.delete_repository.assert_not_called() + + def test_assume_role_failure_raises_actionable_error(self): + from samcli.commands.exceptions import AWSServiceClientError + + role_arn = "arn:aws:iam::123456789012:role/CloudFormationServiceRole" + error = ClientError({"Error": {"Code": "AccessDenied", "Message": "not authorized"}}, "AssumeRole") + self.sts_client.assume_role.side_effect = error + self.boto3_client_mock.side_effect = [ + self.cfn_client, + self.ecr_client, + self.s3_client, + self.sts_client, + Mock(), + ] + + with self.assertRaises(AWSServiceClientError) as ctx: + CompanionStackManager(self.stack_name, "region", "s3_bucket", "s3_prefix", role_arn=role_arn) + self.assertIn(role_arn, str(ctx.exception)) + def test_sync_repos_exists(self): self.manager.does_companion_stack_exist = lambda: True self.manager.get_repository_mapping = lambda: {"a": ""} From 75f5f7cac4ca81c6bb72407ce11588d701e34771 Mon Sep 17 00:00:00 2001 From: Sanjay Santhanam Date: Wed, 9 Sep 2026 12:12:50 -0700 Subject: [PATCH 4/6] fix: lazy non-fatal assume-role for ECR calls; RoleARN on delete_stack --- .../companion_stack_manager.py | 67 ++++++++++++++----- .../test_companion_stack_manager.py | 59 +++++++++++----- 2 files changed, 95 insertions(+), 31 deletions(-) diff --git a/samcli/lib/bootstrap/companion_stack/companion_stack_manager.py b/samcli/lib/bootstrap/companion_stack/companion_stack_manager.py index 71c8cdc7517..60dd35857a1 100644 --- a/samcli/lib/bootstrap/companion_stack/companion_stack_manager.py +++ b/samcli/lib/bootstrap/companion_stack/companion_stack_manager.py @@ -40,6 +40,9 @@ class CompanionStackManager: _s3_prefix: str _cfn_client: CloudFormationClient _s3_client: S3Client + _role_arn: Optional[str] + _sts_client: Any + _role_ecr_client: Any def __init__(self, stack_name, region, s3_bucket, s3_prefix, role_arn: Optional[str] = None): self._companion_stack = CompanionStack(stack_name) @@ -50,18 +53,13 @@ def __init__(self, stack_name, region, s3_bucket, s3_prefix, role_arn: Optional[ self._s3_bucket = s3_bucket self._s3_prefix = s3_prefix self._role_arn = role_arn + self._role_ecr_client = None try: self._cfn_client = boto3.client("cloudformation", config=self._boto_config) self._ecr_client = boto3.client("ecr", config=self._boto_config) self._s3_client = boto3.client("s3", config=self._boto_config) - sts_client = boto3.client("sts") - self._account_id = sts_client.get_caller_identity().get("Account") - if role_arn: - # The CloudFormation RoleARN only covers calls CloudFormation - # makes on our behalf, not SDK calls SAM CLI makes itself - # (e.g. deleting unreferenced ECR repos). Assume the same role - # so all companion-stack side effects use one identity. - self._ecr_client = self._ecr_client_for_role(sts_client, role_arn) + self._sts_client = boto3.client("sts") + self._account_id = self._sts_client.get_caller_identity().get("Account") self._region_name = self._cfn_client.meta.region_name except NoCredentialsError as ex: raise AWSServiceClientError( @@ -112,6 +110,36 @@ def _ecr_client_for_role(self, sts_client, role_arn: str): aws_session_token=credentials["SessionToken"], ) + def _get_ecr_client(self): + """ + Return the ECR client to use for direct ECR calls (e.g. deleting + unreferenced repositories). + + A CloudFormation service role's trust policy generally only trusts + cloudformation.amazonaws.com, so the deploying principal usually + cannot assume it; the assume is therefore attempted lazily — only + when a direct ECR call is actually about to be made — and any failure + falls back to the caller's credentials rather than aborting the + deploy. When no role ARN is configured, the caller's client is used. + + Returns + ------- + An ECR client. + """ + if not self._role_arn: + return self._ecr_client + if self._role_ecr_client is None: + try: + self._role_ecr_client = self._ecr_client_for_role(self._sts_client, self._role_arn) + except AWSServiceClientError as ex: + LOG.debug( + "Unable to assume %s for companion stack ECR operations, " "falling back to caller credentials: %s", + self._role_arn, + ex, + ) + self._role_ecr_client = self._ecr_client + return self._role_ecr_client + def set_functions( self, function_logical_ids: List[str], image_repositories: Optional[Dict[str, str]] = None ) -> None: @@ -185,7 +213,13 @@ def _delete_companion_stack(self) -> None: """ stack_name = self._companion_stack.stack_name waiter = self._cfn_client.get_waiter("stack_delete_complete") - self._cfn_client.delete_stack(StackName=stack_name) + # Use the service role passed via --role-arn (if any) so the delete + # runs under the same identity as create/update; stacks created + # without a role would otherwise fall back to caller credentials. + delete_kwargs: Dict[str, Any] = {"StackName": stack_name} + if self._role_arn: + delete_kwargs["RoleARN"] = self._role_arn + self._cfn_client.delete_stack(**delete_kwargs) waiter.wait(StackName=stack_name, WaiterConfig=self._delete_stack_waiter_config) def list_deployed_repos(self) -> List[ECRRepo]: @@ -240,10 +274,11 @@ def delete_unreferenced_repos(self) -> None: If repo does not exist, this will simply skip it. """ repos = self.get_unreferenced_repos() + ecr_client = self._get_ecr_client() for repo in repos: try: - self._ecr_client.delete_repository(repositoryName=repo.physical_id, force=True) - except self._ecr_client.exceptions.RepositoryNotFoundException: + ecr_client.delete_repository(repositoryName=repo.physical_id, force=True) + except ecr_client.exceptions.RepositoryNotFoundException: LOG.debug("Image repo [%s] not found in companion stack. Skipping deletion.", repo.physical_id) def sync_repos(self) -> None: @@ -353,10 +388,12 @@ def sync_ecr_stack( image_repositories : Dict[str, str] Mapping between function logical ID and ECR URI role_arn : Optional[str] - Optional service role ARN used when creating or updating the companion stack. - When provided, the companion stack is created/updated with the same service - role as the main stack, and direct ECR calls made by SAM CLI (e.g. deleting - unreferenced repositories) assume the role so they run under one identity. + Optional service role ARN used when creating, updating, or deleting + the companion stack. When provided, the companion stack is managed + with the same service role as the main stack, and direct ECR calls + made by SAM CLI (e.g. deleting unreferenced repositories) attempt to + assume the role, falling back to caller credentials if it cannot be + assumed. Returns ------- diff --git a/tests/unit/lib/bootstrap/companion_stack/test_companion_stack_manager.py b/tests/unit/lib/bootstrap/companion_stack/test_companion_stack_manager.py index d2acb98d707..100fabecb67 100644 --- a/tests/unit/lib/bootstrap/companion_stack/test_companion_stack_manager.py +++ b/tests/unit/lib/bootstrap/companion_stack/test_companion_stack_manager.py @@ -132,9 +132,11 @@ def _test_companion_stack_with_role_arn( ] manager = CompanionStackManager(self.stack_name, "region", "s3_bucket", "s3_prefix", role_arn=role_arn) - # the service role is assumed and the ECR client uses its credentials - self.sts_client.assume_role.assert_called_once_with(RoleArn=role_arn, RoleSessionName="sam-cli-companion-stack") - self.assertIs(manager._ecr_client, role_ecr_client) + # construction does not assume the role: the assume is lazy so + # deployments whose service role trust policy only covers + # cloudformation.amazonaws.com keep working. + self.sts_client.assume_role.assert_not_called() + self.assertIs(manager._ecr_client, self.ecr_client) cfn_waiter = Mock() self.cfn_client.get_waiter.return_value = cfn_waiter @@ -290,32 +292,57 @@ def test_delete_unreferenced_repos_with_role_arn_uses_assumed_role_client(self): ] manager = CompanionStackManager(self.stack_name, "region", "s3_bucket", "s3_prefix", role_arn=role_arn) + # the assume is lazy: no STS call until a direct ECR call happens + self.sts_client.assume_role.assert_not_called() + repo = Mock() repo.physical_id = "ECRRepoStale" manager.get_unreferenced_repos = lambda: [repo] manager.delete_unreferenced_repos() + self.sts_client.assume_role.assert_called_once_with(RoleArn=role_arn, RoleSessionName="sam-cli-companion-stack") role_ecr_client.delete_repository.assert_called_once_with(repositoryName="ECRRepoStale", force=True) self.ecr_client.delete_repository.assert_not_called() - def test_assume_role_failure_raises_actionable_error(self): - from samcli.commands.exceptions import AWSServiceClientError - + def test_assume_role_failure_falls_back_to_caller_credentials(self): + # A CloudFormation service role's trust policy generally only trusts + # cloudformation.amazonaws.com, so the deploying principal usually + # cannot assume it. That must not abort the deploy: the assume is + # lazy and a failure falls back to caller credentials. role_arn = "arn:aws:iam::123456789012:role/CloudFormationServiceRole" error = ClientError({"Error": {"Code": "AccessDenied", "Message": "not authorized"}}, "AssumeRole") self.sts_client.assume_role.side_effect = error - self.boto3_client_mock.side_effect = [ - self.cfn_client, - self.ecr_client, - self.s3_client, - self.sts_client, - Mock(), - ] + self.boto3_client_mock.side_effect = [self.cfn_client, self.ecr_client, self.s3_client, self.sts_client] + + # construction succeeds even though the role cannot be assumed + manager = CompanionStackManager(self.stack_name, "region", "s3_bucket", "s3_prefix", role_arn=role_arn) + self.sts_client.assume_role.assert_not_called() - with self.assertRaises(AWSServiceClientError) as ctx: - CompanionStackManager(self.stack_name, "region", "s3_bucket", "s3_prefix", role_arn=role_arn) - self.assertIn(role_arn, str(ctx.exception)) + repo = Mock() + repo.physical_id = "ECRRepoStale" + manager.get_unreferenced_repos = lambda: [repo] + + manager.delete_unreferenced_repos() + manager.delete_unreferenced_repos() + + # the assume was attempted lazily, exactly once, then fell back to + # the caller's credentials + self.sts_client.assume_role.assert_called_once_with(RoleArn=role_arn, RoleSessionName="sam-cli-companion-stack") + self.assertEqual(self.ecr_client.delete_repository.call_count, 2) + + def test_delete_companion_stack_with_role_arn(self): + role_arn = "arn:aws:iam::123456789012:role/CloudFormationServiceRole" + self.boto3_client_mock.side_effect = [self.cfn_client, self.ecr_client, self.s3_client, self.sts_client] + manager = CompanionStackManager(self.stack_name, "region", "s3_bucket", "s3_prefix", role_arn=role_arn) + cfn_waiter = Mock() + self.cfn_client.get_waiter.return_value = cfn_waiter + + manager._delete_companion_stack() + + self.cfn_client.delete_stack.assert_called_once_with(StackName=self.companion_stack_name, RoleARN=role_arn) + self.cfn_client.get_waiter.assert_called_once_with("stack_delete_complete") + cfn_waiter.wait.assert_called_once_with(StackName=self.companion_stack_name, WaiterConfig=ANY) def test_sync_repos_exists(self): self.manager.does_companion_stack_exist = lambda: True From a3353b21f83d3f0b01bf7c7ce749b740aac6f1f5 Mon Sep 17 00:00:00 2001 From: Sanjay Santhanam Date: Wed, 9 Sep 2026 20:49:42 -0700 Subject: [PATCH 5/6] Address review feedback: catch BotoCoreError in _ecr_client_for_role and skip AssumeRole when there are no stale repos --- .../companion_stack_manager.py | 13 ++++++- .../test_companion_stack_manager.py | 37 +++++++++++++++++++ 2 files changed, 48 insertions(+), 2 deletions(-) diff --git a/samcli/lib/bootstrap/companion_stack/companion_stack_manager.py b/samcli/lib/bootstrap/companion_stack/companion_stack_manager.py index 60dd35857a1..4ffc733e17e 100644 --- a/samcli/lib/bootstrap/companion_stack/companion_stack_manager.py +++ b/samcli/lib/bootstrap/companion_stack/companion_stack_manager.py @@ -7,7 +7,7 @@ import boto3 from botocore.config import Config -from botocore.exceptions import ClientError, NoCredentialsError, NoRegionError +from botocore.exceptions import BotoCoreError, ClientError, NoCredentialsError, NoRegionError from mypy_boto3_cloudformation.client import CloudFormationClient from mypy_boto3_cloudformation.type_defs import WaiterConfigTypeDef from mypy_boto3_s3.client import S3Client @@ -97,7 +97,11 @@ def _ecr_client_for_role(self, sts_client, role_arn: str): """ try: assumed_role = sts_client.assume_role(RoleArn=role_arn, RoleSessionName="sam-cli-companion-stack") - except ClientError as ex: + except (ClientError, BotoCoreError) as ex: + # BotoCoreError covers ParamValidationError (e.g. a typo'd role + # ARN, validated client-side), NoCredentialsError and + # endpoint/connection errors, so every assume_role failure becomes + # an AWSServiceClientError that _get_ecr_client can fall back from. raise AWSServiceClientError( f"Error assuming the provided role {role_arn} for companion stack ECR operations: {ex}" ) from ex @@ -274,6 +278,11 @@ def delete_unreferenced_repos(self) -> None: If repo does not exist, this will simply skip it. """ repos = self.get_unreferenced_repos() + if not repos: + # Nothing to delete: skip the (lazy) assume-role attempt so a + # routine deploy with --role-arn does not emit a pointless + # sts:AssumeRole call. + return ecr_client = self._get_ecr_client() for repo in repos: try: diff --git a/tests/unit/lib/bootstrap/companion_stack/test_companion_stack_manager.py b/tests/unit/lib/bootstrap/companion_stack/test_companion_stack_manager.py index 100fabecb67..946f4cac8fb 100644 --- a/tests/unit/lib/bootstrap/companion_stack/test_companion_stack_manager.py +++ b/tests/unit/lib/bootstrap/companion_stack/test_companion_stack_manager.py @@ -331,6 +331,43 @@ def test_assume_role_failure_falls_back_to_caller_credentials(self): self.sts_client.assume_role.assert_called_once_with(RoleArn=role_arn, RoleSessionName="sam-cli-companion-stack") self.assertEqual(self.ecr_client.delete_repository.call_count, 2) + def test_assume_role_param_validation_error_falls_back_to_caller_credentials(self): + # --role-arn is an unvalidated free-form string; botocore validates it + # client-side and raises ParamValidationError (a BotoCoreError, not a + # ClientError). That must also translate into the documented fallback + # to caller credentials instead of aborting the deploy with a raw + # botocore traceback. + from botocore.exceptions import ParamValidationError + + role_arn = "not-an-arn" + self.sts_client.assume_role.side_effect = ParamValidationError(report="Invalid RoleArn") + self.boto3_client_mock.side_effect = [self.cfn_client, self.ecr_client, self.s3_client, self.sts_client] + + manager = CompanionStackManager(self.stack_name, "region", "s3_bucket", "s3_prefix", role_arn=role_arn) + + repo = Mock() + repo.physical_id = "ECRRepoStale" + manager.get_unreferenced_repos = lambda: [repo] + + manager.delete_unreferenced_repos() + + self.sts_client.assume_role.assert_called_once_with(RoleArn=role_arn, RoleSessionName="sam-cli-companion-stack") + self.ecr_client.delete_repository.assert_called_once_with(repositoryName="ECRRepoStale", force=True) + + def test_delete_unreferenced_repos_without_stale_repos_skips_assume_role(self): + # Zero stale repos is the steady state; the lazy assume-role attempt + # must not fire (and must not appear in CloudTrail) when there is + # nothing to delete. + role_arn = "arn:aws:iam::123456789012:role/CloudFormationServiceRole" + self.boto3_client_mock.side_effect = [self.cfn_client, self.ecr_client, self.s3_client, self.sts_client] + manager = CompanionStackManager(self.stack_name, "region", "s3_bucket", "s3_prefix", role_arn=role_arn) + manager.get_unreferenced_repos = lambda: [] + + manager.delete_unreferenced_repos() + + self.sts_client.assume_role.assert_not_called() + self.ecr_client.delete_repository.assert_not_called() + def test_delete_companion_stack_with_role_arn(self): role_arn = "arn:aws:iam::123456789012:role/CloudFormationServiceRole" self.boto3_client_mock.side_effect = [self.cfn_client, self.ecr_client, self.s3_client, self.sts_client] From c04273db13ef4e5370b7ca5be714a7a1421e37d6 Mon Sep 17 00:00:00 2001 From: Sanjay Santhanam Date: Wed, 9 Sep 2026 21:36:21 -0700 Subject: [PATCH 6/6] Address review: complete ECR fallback contract for assumed service role --- .../companion_stack_manager.py | 40 +++++--- .../test_companion_stack_manager.py | 91 +++++++++++++++++++ 2 files changed, 120 insertions(+), 11 deletions(-) diff --git a/samcli/lib/bootstrap/companion_stack/companion_stack_manager.py b/samcli/lib/bootstrap/companion_stack/companion_stack_manager.py index 4ffc733e17e..7633d6232d0 100644 --- a/samcli/lib/bootstrap/companion_stack/companion_stack_manager.py +++ b/samcli/lib/bootstrap/companion_stack/companion_stack_manager.py @@ -97,22 +97,25 @@ def _ecr_client_for_role(self, sts_client, role_arn: str): """ try: assumed_role = sts_client.assume_role(RoleArn=role_arn, RoleSessionName="sam-cli-companion-stack") - except (ClientError, BotoCoreError) as ex: + credentials = assumed_role["Credentials"] + return boto3.client( + "ecr", + config=self._boto_config, + aws_access_key_id=credentials["AccessKeyId"], + aws_secret_access_key=credentials["SecretAccessKey"], + aws_session_token=credentials["SessionToken"], + ) + except (ClientError, BotoCoreError, KeyError) as ex: # BotoCoreError covers ParamValidationError (e.g. a typo'd role # ARN, validated client-side), NoCredentialsError and - # endpoint/connection errors, so every assume_role failure becomes - # an AWSServiceClientError that _get_ecr_client can fall back from. + # endpoint/connection errors; KeyError covers a malformed assume + # response. The response-consumption statements live inside the + # try as well, so every failure here becomes an + # AWSServiceClientError that _get_ecr_client can fall back from + # instead of aborting the deploy. raise AWSServiceClientError( f"Error assuming the provided role {role_arn} for companion stack ECR operations: {ex}" ) from ex - credentials = assumed_role["Credentials"] - return boto3.client( - "ecr", - config=self._boto_config, - aws_access_key_id=credentials["AccessKeyId"], - aws_secret_access_key=credentials["SecretAccessKey"], - aws_session_token=credentials["SessionToken"], - ) def _get_ecr_client(self): """ @@ -289,6 +292,21 @@ def delete_unreferenced_repos(self) -> None: ecr_client.delete_repository(repositoryName=repo.physical_id, force=True) except ecr_client.exceptions.RepositoryNotFoundException: LOG.debug("Image repo [%s] not found in companion stack. Skipping deletion.", repo.physical_id) + except ClientError as ex: + # A CloudFormation service role is commonly scoped to what CFN + # needs to create/update the stack; if it can be assumed but + # lacks ecr:DeleteRepository, the AccessDenied ClientError is + # unhandled above and would abort the deploy. Since the + # assumed role is best-effort, retry with the caller's + # credentials instead of failing. + error_code = ex.response.get("Error", {}).get("Code") + if ecr_client is self._ecr_client or error_code != "AccessDeniedException": + raise + LOG.debug( + "Assumed role not authorized to delete image repo [%s]; retrying with caller credentials.", + repo.physical_id, + ) + self._ecr_client.delete_repository(repositoryName=repo.physical_id, force=True) def sync_repos(self) -> None: """ diff --git a/tests/unit/lib/bootstrap/companion_stack/test_companion_stack_manager.py b/tests/unit/lib/bootstrap/companion_stack/test_companion_stack_manager.py index 946f4cac8fb..3b4acb1e023 100644 --- a/tests/unit/lib/bootstrap/companion_stack/test_companion_stack_manager.py +++ b/tests/unit/lib/bootstrap/companion_stack/test_companion_stack_manager.py @@ -368,6 +368,97 @@ def test_delete_unreferenced_repos_without_stale_repos_skips_assume_role(self): self.sts_client.assume_role.assert_not_called() self.ecr_client.delete_repository.assert_not_called() + def test_malformed_assume_role_response_falls_back_to_caller_credentials(self): + # A malformed AssumeRole response (e.g. a Credentials dict missing + # the access keys) raises KeyError when the response is consumed. + # That must translate into the documented fallback to caller + # credentials, not abort the deploy. + role_arn = "arn:aws:iam::123456789012:role/CloudFormationServiceRole" + self.sts_client.assume_role.return_value = {"Credentials": {}} + self.boto3_client_mock.side_effect = [self.cfn_client, self.ecr_client, self.s3_client, self.sts_client] + + manager = CompanionStackManager(self.stack_name, "region", "s3_bucket", "s3_prefix", role_arn=role_arn) + + repo = Mock() + repo.physical_id = "ECRRepoStale" + manager.get_unreferenced_repos = lambda: [repo] + + manager.delete_unreferenced_repos() + + self.sts_client.assume_role.assert_called_once_with(RoleArn=role_arn, RoleSessionName="sam-cli-companion-stack") + self.ecr_client.delete_repository.assert_called_once_with(repositoryName="ECRRepoStale", force=True) + + def test_role_client_access_denied_on_delete_retries_with_caller_credentials(self): + # A CloudFormation service role may be assumable yet lack + # ecr:DeleteRepository. The resulting AccessDenied ClientError on the + # assumed-role client must fall back to the caller's client instead + # of aborting the deploy. + role_arn = "arn:aws:iam::123456789012:role/CloudFormationServiceRole" + role_ecr_client = Mock() + role_ecr_client.exceptions.RepositoryNotFoundException = type("RepositoryNotFoundException", (Exception,), {}) + access_denied = ClientError( + {"Error": {"Code": "AccessDeniedException", "Message": "not authorized"}}, "DeleteRepository" + ) + role_ecr_client.delete_repository.side_effect = access_denied + self.sts_client.assume_role.return_value = { + "Credentials": { + "AccessKeyId": "", + "SecretAccessKey": "secret", + "SessionToken": "token", + } + } + self.boto3_client_mock.side_effect = [ + self.cfn_client, + self.ecr_client, + self.s3_client, + self.sts_client, + role_ecr_client, + ] + manager = CompanionStackManager(self.stack_name, "region", "s3_bucket", "s3_prefix", role_arn=role_arn) + + repo = Mock() + repo.physical_id = "ECRRepoStale" + manager.get_unreferenced_repos = lambda: [repo] + + manager.delete_unreferenced_repos() + + role_ecr_client.delete_repository.assert_called_once_with(repositoryName="ECRRepoStale", force=True) + self.ecr_client.delete_repository.assert_called_once_with(repositoryName="ECRRepoStale", force=True) + + def test_role_client_non_access_denied_error_still_raises(self): + # Only the AccessDenied permission failure deserves the caller- + # credential retry; any other ClientError from the assumed-role + # client must keep propagating. + role_arn = "arn:aws:iam::123456789012:role/CloudFormationServiceRole" + role_ecr_client = Mock() + role_ecr_client.exceptions.RepositoryNotFoundException = type("RepositoryNotFoundException", (Exception,), {}) + role_ecr_client.delete_repository.side_effect = ClientError( + {"Error": {"Code": "InvalidParameterException", "Message": "bad param"}}, "DeleteRepository" + ) + self.sts_client.assume_role.return_value = { + "Credentials": { + "AccessKeyId": "", + "SecretAccessKey": "secret", + "SessionToken": "token", + } + } + self.boto3_client_mock.side_effect = [ + self.cfn_client, + self.ecr_client, + self.s3_client, + self.sts_client, + role_ecr_client, + ] + manager = CompanionStackManager(self.stack_name, "region", "s3_bucket", "s3_prefix", role_arn=role_arn) + + repo = Mock() + repo.physical_id = "ECRRepoStale" + manager.get_unreferenced_repos = lambda: [repo] + + with self.assertRaises(ClientError): + manager.delete_unreferenced_repos() + self.ecr_client.delete_repository.assert_not_called() + def test_delete_companion_stack_with_role_arn(self): role_arn = "arn:aws:iam::123456789012:role/CloudFormationServiceRole" self.boto3_client_mock.side_effect = [self.cfn_client, self.ecr_client, self.s3_client, self.sts_client]