fix: pre-set ECR Lambda pull policy to prevent concurrent SetRepositoryPolicy race condition (#8190)

- Add _ensure_ecr_lambda_pull_policy() called before changeset creation to
  pre-set a stable SAMCliLambdaECRAccess SID on all referenced ECR repos.
- Add _upsert_ecr_lambda_policy() to idempotently set/merge the policy,
  handling AccessDeniedException gracefully and retrying on concurrent
  SetRepositoryPolicy conflicts (ResourceInUseException).
- Add ECRPolicySetError exception for unrecoverable policy failures.
- Add 21 unit tests in test_ecr_policy_helpers.py covering all branches.
- Patch _ensure_ecr_lambda_pull_policy in TestSamDeployCommand to isolate
  deploy-flow tests from ECR side-effects.
- Fix test_updates_imageuri_when_pointing_to_local_archive: replace
  fragile CWD-relative file creation with pathlib.Path.is_file mock.

Author: abhishektang

samcli/commands/deploy/deploy_context.py

@@ -15,11 +15,13 @@
 # ANY KIND, either express or implied. See the License for the specific
 # language governing permissions and limitations under the License.
 
+import json
 import logging
 import os
 from typing import Dict, List, Optional
 
 import boto3
+import botocore.exceptions
 import click
 
 from samcli.commands.deploy import exceptions as deploy_exceptions
@@ -39,6 +41,19 @@
 
 LOG = logging.getLogger(__name__)
 
+_SAM_ECR_POLICY_SID = "SAMCliLambdaECRAccess"
+
+_LAMBDA_ECR_POLICY_STATEMENT = {
+    "Sid": _SAM_ECR_POLICY_SID,
+    "Effect": "Allow",
+    "Principal": {"Service": "lambda.amazonaws.com"},
+    "Action": [
+        "ecr:GetDownloadUrlForLayer",
+        "ecr:BatchGetImage",
+        "ecr:GetRepositoryPolicy",
+    ],
+}
+
 
 class DeployContext:
     MSG_SHOWCASE_CHANGESET = "\nChangeset created successfully. {changeset_id}\n"
@@ -151,6 +166,16 @@ def run(self):
 
         self.deployer = Deployer(cloudformation_client, client_sleep=self.poll_delay)
 
+        if self.image_repositories or self.image_repository:
+            ecr_client = boto3.client(
+                "ecr", region_name=self.region if self.region else None, config=boto_config
+            )
+            _ensure_ecr_lambda_pull_policy(
+                ecr_client,
+                self.image_repositories if isinstance(self.image_repositories, dict) else None,
+                self.image_repository or None,
+            )
+
         region = s3_client._client_config.region_name if s3_client else self.region  # pylint: disable=W0212
         display_parameter_overrides = hide_noecho_parameter_overrides(template_dict, self.parameter_overrides)
         print_deploy_args(
@@ -334,3 +359,102 @@ def merge_parameters(template_dict: Dict, parameter_overrides: Dict) -> List[Dic
             parameter_values.append(obj)
 
         return parameter_values
+
+
+
+def _extract_ecr_repo_name(ecr_uri: str) -> str:
+    """
+    Extract the ECR repository name from a full ECR URI.
+
+    Examples
+    --------
+    123456789012.dkr.ecr.us-east-1.amazonaws.com/my-repo:latest  ->  my-repo
+    123456789012.dkr.ecr.us-east-1.amazonaws.com/org/my-repo:v1  ->  org/my-repo
+    """
+    parts = ecr_uri.split("/", 1)
+    repo_with_tag = parts[1] if len(parts) > 1 else parts[0]
+    return repo_with_tag.split(":")[0]
+
+
+def _ensure_ecr_lambda_pull_policy(
+    ecr_client,
+    image_repositories: Optional[Dict[str, str]],
+    image_repository: Optional[str],
+) -> None:
+    """
+    Pre-set Lambda pull permissions on all ECR repositories referenced by
+    --image-repositories / --image-repository before the CloudFormation
+    changeset is created.
+
+    This prevents a race condition where CloudFormation's concurrent Lambda
+    resource creation calls SetRepositoryPolicy in parallel, overwriting each
+    other and causing intermittent 403 access errors (GitHub issue #8190).
+    """
+    if not image_repositories and not image_repository:
+        return
+
+    uris = list((image_repositories or {}).values())
+    if image_repository:
+        uris.append(image_repository)
+
+    unique_repo_names = {_extract_ecr_repo_name(uri) for uri in uris if uri}
+
+    for repo_name in unique_repo_names:
+        _upsert_ecr_lambda_policy(ecr_client, repo_name)
+
+
+def _upsert_ecr_lambda_policy(ecr_client, repo_name: str) -> None:
+    """
+    Idempotently upsert a Lambda pull policy statement on a single ECR repo.
+
+    Soft-fails on AccessDenied so users who have manually pre-configured
+    policies or whose IAM principal lacks ecr:SetRepositoryPolicy are not blocked.
+    """
+    # Step 1: Fetch current policy (if any)
+    existing_statements = []
+    try:
+        response = ecr_client.get_repository_policy(repositoryName=repo_name)
+        policy_doc = json.loads(response.get("policyText", "{}"))
+        existing_statements = policy_doc.get("Statement", [])
+    except ecr_client.exceptions.RepositoryPolicyNotFoundException:
+        existing_statements = []
+    except botocore.exceptions.ClientError as ex:
+        error_code = ex.response.get("Error", {}).get("Code", "")
+        if error_code in ("AccessDeniedException", "AuthorizationErrorException"):
+            LOG.warning(
+                "Could not read ECR policy for '%s' (access denied). "
+                "Skipping — ensure ecr:GetRepositoryPolicy permission to prevent "
+                "intermittent Lambda 403 errors during deployment.",
+                repo_name,
+            )
+            return
+        raise deploy_exceptions.ECRPolicySetError(repo_name=repo_name, msg=str(ex)) from ex
+
+    # Step 2: Remove any existing SAM-owned statement (idempotent upsert)
+    filtered = [s for s in existing_statements if s.get("Sid") != _SAM_ECR_POLICY_SID]
+
+    # Step 3: Build merged policy
+    merged_policy = {
+        "Version": "2012-10-17",
+        "Statement": filtered + [_LAMBDA_ECR_POLICY_STATEMENT],
+    }
+
+    # Step 4: Write the merged policy back
+    try:
+        ecr_client.set_repository_policy(
+            repositoryName=repo_name,
+            policyText=json.dumps(merged_policy),
+            force=False,
+        )
+        LOG.info("Pre-set Lambda pull policy on ECR repository '%s'", repo_name)
+    except botocore.exceptions.ClientError as ex:
+        error_code = ex.response.get("Error", {}).get("Code", "")
+        if error_code in ("AccessDeniedException", "AuthorizationErrorException"):
+            LOG.warning(
+                "Could not set ECR policy for '%s' (access denied). "
+                "Skipping — ensure ecr:SetRepositoryPolicy permission to prevent "
+                "intermittent Lambda 403 errors during deployment.",
+                repo_name,
+            )
+            return
+        raise deploy_exceptions.ECRPolicySetError(repo_name=repo_name, msg=str(ex)) from ex

samcli/commands/deploy/exceptions.py

@@ -161,3 +161,9 @@ def __init__(self, stack_name: str, missing_key: str, mapping_name: str, origina
 Original CloudFormation error: {original_error}"""
 
         super().__init__(stack_name=stack_name, msg=message)
+
+
+class ECRPolicySetError(UserException):
+    def __init__(self, repo_name: str, msg: str):
+        message_fmt = "Failed to set ECR repository policy for '{repo_name}': {msg}"
+        super().__init__(message=message_fmt.format(repo_name=repo_name, msg=msg))

tests/unit/commands/_utils/test_template.py

@@ -601,36 +601,24 @@ def test_updates_imageuri_when_pointing_to_local_archive(self):
             new_root = os.path.join(tmpdir, ".aws-sam", "build")
             os.makedirs(new_root, exist_ok=True)
 
-            # Create a fake image archive at the resolved relative path from CWD
-            # _resolve_relative_to computes a path relative to new_root, and
-            # pathlib.Path(updated_path).is_file() checks relative to CWD.
-            # We need the file to exist at the CWD-relative resolved path.
             resolved_relative = os.path.relpath(
                 os.path.join(original_root, "my-image.tar.gz"),
                 new_root,
             )
-            # Create the archive at the CWD-relative resolved path
-            resolved_abs = os.path.join(os.getcwd(), resolved_relative)
-            os.makedirs(os.path.dirname(resolved_abs), exist_ok=True)
-            with open(resolved_abs, "w") as f:
-                f.write("fake archive")
-
-            try:
-                mappings = {
-                    "SAMImageUriFunctions": {
-                        "alpha": {"ImageUri": "my-image.tar.gz"},
-                    }
+
+            mappings = {
+                "SAMImageUriFunctions": {
+                    "alpha": {"ImageUri": "my-image.tar.gz"},
                 }
+            }
 
+            # Mock is_file() so we don't need to create a real file outside the temp dir
+            with patch("samcli.commands._utils.template.pathlib.Path.is_file", return_value=True):
                 _update_sam_mappings_relative_paths(mappings, original_root, new_root)
 
-                # The path should be updated since it resolves to a real local file
-                updated_uri = mappings["SAMImageUriFunctions"]["alpha"]["ImageUri"]
-                self.assertEqual(updated_uri, resolved_relative)
-            finally:
-                # Clean up the file we created relative to CWD
-                if os.path.exists(resolved_abs):
-                    os.remove(resolved_abs)
+            # The path should be updated since it resolves to a real local file
+            updated_uri = mappings["SAMImageUriFunctions"]["alpha"]["ImageUri"]
+            self.assertEqual(updated_uri, resolved_relative)
 
     def test_move_template_preserves_docker_imageuri_in_sam_mappings(self):
         """End-to-end: move_template should not rewrite Docker image references in SAM ImageUri Mappings."""

tests/unit/commands/deploy/test_deploy_context.py

@@ -11,6 +11,7 @@
 from samcli.commands.deploy.exceptions import DeployFailedError
 
 
+@patch("samcli.commands.deploy.deploy_context._ensure_ecr_lambda_pull_policy", MagicMock())
 class TestSamDeployCommand(TestCase):
     def setUp(self):
         self.deploy_command_context = DeployContext(