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 | ||
|
|
@@ -41,19 +41,27 @@ class CompanionStackManager: | |
| _cfn_client: CloudFormationClient | ||
| _s3_client: S3Client | ||
|
|
||
| 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 | ||
| 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: | ||
|
There was a problem hiding this comment. [BUG] The constructor now calls
The assume also happens eagerly even when no direct ECR call will ever be made (the only use of Suggested direction: keep the def getecr_client(self):
"""Return the assumed-role ECR client when available, else the caller's 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(boto3.client("sts"), self._role_arn)
except AWSServiceClientError as ex:
LOG.debug("Unable to assume %s for ECR operations, using caller credentials: %s", self._role_arn, ex)
self._role_ecr_client = self._ecr_client
return self._role_ecr_clientIf assuming the service role is genuinely intended as a hard requirement, that is a user-visible behavior change that needs to be called out in the PR description and documentation, not just an internal detail.
Author
There was a problem hiding this comment. Good call — that was a real regression I introduced. The eager assume is gone from the constructor; the role is now assumed lazily only when a direct ECR call is actually about to happen, and an assume failure just falls back to caller credentials with a debug log instead of aborting the deploy. When no --role-arn is set, nothing changed. CloudFormation create/update/delete keep their RoleARN wiring. Reworked the tests: assume is attempted lazily and exactly once, the failure case now asserts construction succeeds and ECR deletes run under caller creds, plus a case that no STS call happens when there are no stale repos. |
||
| # 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( | ||
|
|
@@ -68,6 +76,42 @@ 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 set_functions( | ||
| self, function_logical_ids: List[str], image_repositories: Optional[Dict[str, str]] = None | ||
| ) -> None: | ||
|
|
@@ -117,16 +161,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) | ||
|
|
||
|
|
@@ -279,7 +328,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 +352,11 @@ 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, and direct ECR calls made by SAM CLI (e.g. deleting | ||
| unreferenced repositories) assume the role so they run under one identity. | ||
|
|
||
| Returns | ||
| ------- | ||
|
|
@@ -305,7 +365,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.