fix: use --role-arn service role for companion stack create/update #9267
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -3,7 +3,7 @@ | |
| """ | ||
|
|
||
| import logging | ||
| from typing import Any, Dict, List, Optional | ||
|
|
||
| import boto3 | ||
| from botocore.config import Config | ||
|
|
@@ -40,20 +40,26 @@ 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) | ||
| self._builder = CompanionStackBuilder(self._companion_stack) | ||
| self._boto_config = Config(region_name=region if region else None) | ||
| self._update_stack_waiter_config = {"Delay": 10, "MaxAttempts": 120} | ||
| self._delete_stack_waiter_config = {"Delay": 10, "MaxAttempts": 120} | ||
| 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) | ||
| 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( | ||
|
|
@@ -68,6 +74,72 @@ def __init__(self, stack_name, region, s3_bucket, s3_prefix): | |
| "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: | ||
|
There was a problem hiding this comment. [ERROR_HANDLING] The This is reachable without an unusual setup: Widening the caught type keeps the documented behavior intact: from botocore.exceptions import BotoCoreError, ClientError
try:
assumed_role = sts_client.assume_role(RoleArn=role_arn, RoleSessionName="sam-cli-companion-stack")
except (ClientError, BotoCoreError) as ex:
raise AWSServiceClientError(
f"Error assuming the provided role {role_arn} for companion stack ECR operations: {ex}"
) from ex
Author
There was a problem hiding this comment. Good point — a malformed role ARN fails with ParamValidationError before any ClientError is raised, so the fallback would never trigger there. |
||
| raise AWSServiceClientError( | ||
| f"Error assuming the provided role {role_arn} for companion stack ECR operations: {ex}" | ||
| ) from ex | ||
| credentials = assumed_role["Credentials"] | ||
|
There was a problem hiding this comment. [ERROR_HANDLING] The except (ClientError, BotoCoreError) as ex:
raise AWSServiceClientError(...) from ex
credentials = assumed_role["Credentials"] # outside try
return boto3.client( # outside try
"ecr",
config=self._boto_config,
aws_access_key_id=credentials["AccessKeyId"],
...
)
try:
assumed_role = sts_client.assume_role(RoleArn=role_arn, RoleSessionName="sam-cli-companion-stack")
credentials = assumed_role["Credentials"]
return boto3.client(
"ecr",
config=self._boto_config,
aws_access_key_id=credentials["AccessKeyId"],
aws_secret_access_[REDACTED]
aws_session_[REDACTED]
)
except (ClientError, BotoCoreError, KeyError) as ex:
raise AWSServiceClientError(
f"Error assuming the provided role {role_arn} for companion stack ECR operations: {ex}"
) from exThis also makes the
Author
There was a problem hiding this comment. Fixed — the try block now also covers credential extraction and ECR client construction, so any failure there (malformed AssumeRole response, client construction errors) raises AWSServiceClientError and falls back to the caller-credential client. |
||
| 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): | ||
| """ | ||
| 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: | ||
|
|
@@ -117,16 +189,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 | ||
|
There was a problem hiding this comment. [BUG] The service role is now wired into def sync_repos(self) -> None:
has_repo = bool(self.get_repository_mapping())
if self.does_companion_stack_exist():
self.delete_unreferenced_repos() # <-- direct ECR call, caller creds
...def delete_unreferenced_repos(self) -> None:
repos = self.get_unreferenced_repos()
for repo in repos:
try:
self._ecr_client.delete_repository(repositoryName=repo.physical_id, force=True)
except self._ecr_client.exceptions.RepositoryNotFoundException:
LOG.debug(...)
Two options, either is fine:
Worth covering with a unit test for the unreferenced-repo path, since the current tests exercise only create/update.
Author
There was a problem hiding this comment. Good catch — when role_arn is set, the manager now assumes it via STS and builds self._ecr_client from the assumed-role credentials, so delete_unreferenced_repos and all other direct ECR calls run under one identity. An assume-role failure raises an actionable AWSServiceClientError. Added tests covering the assumed-role delete path and the failure case. There was a problem hiding this comment. [BUG] The service role is applied to if self.does_companion_stack_exist():
self.delete_unreferenced_repos()
if has_repo:
self.update_companion_stack()
else:
self._delete_companion_stack() # no RoleARN
def deletecompanion_stack(self) -> None:
stack_name = self._companion_stack.stack_name
waiter = self._cfn_client.get_waiter("stack_delete_complete")
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)
Author
There was a problem hiding this comment. Fixed — _delete_companion_stack now passes RoleARN on delete_stack when the role is set, so the whole companion-stack lifecycle (create/update/delete) runs under one identity. Also covered by a new test asserting delete_stack gets the RoleARN kwarg; the no-role case still deletes with just StackName. |
||
| if exists: | ||
| 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(**stack_kwargs) | ||
| create_waiter = self._cfn_client.get_waiter("stack_create_complete") | ||
| create_waiter.wait(StackName=stack_name, WaiterConfig=self._update_stack_waiter_config) | ||
|
|
||
|
|
@@ -136,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") | ||
| # 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]: | ||
|
|
@@ -191,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() | ||
|
There was a problem hiding this comment. [GENERAL] The empty case is the steady state, not an edge case: Guarding on the repo list resolves it: repos = self.get_unreferenced_repos()
if not repos:
return
ecr_client = self._get_ecr_client()
for repo in repos:
...Note on prior review threads: the four earlier findings (guided-path plumbing, RoleARN on _delete_companion_stack, the unconditional constructor-time assume_role, and the direct-ECR identity mismatch) are all resolved in this diff and were not re-raised.
Author
There was a problem hiding this comment. Fixed — |
||
| for repo in repos: | ||
| try: | ||
| ecr_client.delete_repository(repositoryName=repo.physical_id, force=True) | ||
| except ecr_client.exceptions.RepositoryNotFoundException: | ||
|
There was a problem hiding this comment. [BUG] Routing Note this is reachable on ordinary deploys, not just teardown: Since the design already treats the assumed role as best-effort, the permission failure deserves the same treatment — fall back to the caller's client rather than failing the deploy: for repo in repos:
try:
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:
if ecr_client is self._ecr_client or ex.response.get("Error", {}).get("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)Re-raising for any other error code keeps genuine failures visible instead of silently swallowing them.
Author
There was a problem hiding this comment. Fixed — if the assumed-role delete hits AccessDeniedException, we now retry the delete with the caller-credential client instead of aborting the deploy. Other errors still propagate. |
||
| LOG.debug("Image repo [%s] not found in companion stack. Skipping deletion.", repo.physical_id) | ||
|
|
||
| def sync_repos(self) -> None: | ||
|
|
@@ -279,7 +363,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], | ||
| role_arn: Optional[str] = None, | ||
| ) -> Dict[str, str]: | ||
| """Blocking call to sync local functions with ECR Companion Stack | ||
|
|
||
|
|
@@ -297,6 +387,13 @@ 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, 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 | ||
| ------- | ||
|
|
@@ -305,7 +402,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, role_arn=role_arn) | ||
|
|
||
| stacks = SamLocalStackProvider.get_stacks(template_file, language_extensions_enabled=False)[0] | ||
| function_provider = SamFunctionProvider(stacks, ignore_code_extraction_warnings=True) | ||
|
|
||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
[BUG] The fix only covers the non-guided branch, but
--role-arnapplies to guided deploys too — it is passed toDeployContextunconditionally atcommand.py:416, and it is also persisted intosamconfig.toml. That meanssam deploy --guided --role-arn <role> --resolve-image-reposstill reproduces the exact failure from #5051: the main stack deploys under the service role, while the companion stack is created with the caller's credentials and fails onecr:CreateRepository.Both companion-stack entry points in the guided flow are affected:
samcli/commands/deploy/guided_context.py:191—sync_ecr_stack(...)when--resolve-image-reposis setsamcli/commands/deploy/guided_context.py:362—CompanionStackManager(stack_name, region, s3_bucket, s3_prefix)inprompt_image_repository, which callsmanager.sync_repos()at line 390 and therefore reachesupdate_companion_stack()/create_stackThreading
role_arnthroughGuidedContextwould close the gap, e.g. accept it inGuidedContext.__init__, passrole_arn=role_arnfrom theguidedbranch ofdo_cli, and forward it at both call sites:If leaving guided out is deliberate, it is worth stating why in the PR description, since the reported symptom is identical in that path.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Good catch — you're right, the guided path had the same bug.
GuidedContextnow acceptsrole_arn,do_clipasses--role-arnthrough on the guided branch, and it's forwarded at both companion-stack call sites (sync_ecr_stackinguided_promptsand theCompanionStackManagerinprompt_image_repository). Added guided tests asserting the role reaches both. Verified: 24/24 guided-context tests and all companion-stack manager tests pass.