ECR: implement layer upload APIs

moto/ecr/exceptions.py

@@ -122,3 +122,15 @@ def __init__(self, image_id: str, repository_name: str, registry_id: str):
 
 class ValidationException(ServiceException):
     code = "ValidationException"
+
+
+class UploadNotFoundException(ServiceException):
+    code = "UploadNotFoundException"
+
+    def __init__(self, upload_id: str, repository_name: str, registry_id: str):
+        message = (
+            f"Upload with id '{upload_id}' could not be found "
+            f"for the repository with name '{repository_name}' "
+            f"in the registry with id '{registry_id}'"
+        )
+        super().__init__(message)

moto/ecr/models.py

@@ -27,6 +27,7 @@
     RepositoryNotFoundException,
     RepositoryPolicyNotFoundException,
     ScanNotFoundException,
+    UploadNotFoundException,
     ValidationException,
 )
 from moto.ecr.policy_validation import EcrLifecyclePolicyValidator
@@ -47,6 +48,12 @@
 
 ImageTagMutabilityExclusionFilterT = dict[Literal["filter", "filterType"], str]
 
+# Advisory part size (in bytes) returned by InitiateLayerUpload. Amazon ECR does
+# not strictly enforce this value, so moto uses a fixed default.
+DEFAULT_LAYER_PART_SIZE = 10 * 1024 * 1024
+# Media type reported for image layers by BatchCheckLayerAvailability.
+LAYER_MEDIA_TYPE = "application/vnd.docker.image.rootfs.diff.tar.gzip"
+
 
 class RepoTagMutability(str, Enum):
     MUTABLE = "MUTABLE"
@@ -97,6 +104,10 @@ def __init__(
         self.policy: str | None = None
         self.lifecycle_policy: str | None = None
         self.images: list[Image] = []
+        # Completed image layers, keyed by digest -> size in bytes.
+        self.layers: dict[str, int] = {}
+        # In-progress layer uploads, keyed by uploadId.
+        self.uploads: dict[str, LayerUpload] = {}
         self.scanning_config = {
             "repositoryArn": self.arn,
             "repositoryName": self.name,
@@ -432,6 +443,20 @@ def update_tag(self, tag: str) -> None:
             self.image_tags.append(tag)
 
 
+class LayerUpload(BaseModel):
+    def __init__(self, repository_name: str, registry_id: str):
+        self.upload_id = str(random.uuid4())
+        self.repository_name = repository_name
+        self.registry_id = registry_id
+        self.layer_parts = bytearray()
+
+    @property
+    def last_byte_received(self) -> int:
+        # Byte index of the last received byte. -1 when nothing has been
+        # uploaded yet, matching how Amazon ECR reports contiguous uploads.
+        return len(self.layer_parts) - 1
+
+
 class ECRBackend(BaseBackend):
     def __init__(self, region_name: str, account_id: str):
         super().__init__(region_name, account_id)
@@ -807,6 +832,99 @@ def _find_images_with_tags_matching_exclusion_filters(
             )
         )
 
+    def initiate_layer_upload(
+        self, repository_name: str, registry_id: str | None = None
+    ) -> dict[str, Any]:
+        repository = self._get_repository(repository_name, registry_id)
+        upload = LayerUpload(repository_name, repository.registry_id)
+        repository.uploads[upload.upload_id] = upload
+        return {"uploadId": upload.upload_id, "partSize": DEFAULT_LAYER_PART_SIZE}
+
+    def upload_layer_part(
+        self,
+        repository_name: str,
+        upload_id: str,
+        part_first_byte: int,
+        part_last_byte: int,
+        layer_part_blob: bytes,
+        registry_id: str | None = None,
+    ) -> dict[str, Any]:
+        repository = self._get_repository(repository_name, registry_id)
+        upload = repository.uploads.get(upload_id)
+        if upload is None:
+            raise UploadNotFoundException(
+                upload_id, repository_name, repository.registry_id
+            )
+        upload.layer_parts += layer_part_blob or b""
+        return {
+            "registryId": repository.registry_id,
+            "repositoryName": repository_name,
+            "uploadId": upload_id,
+            "lastByteReceived": upload.last_byte_received,
+        }
+
+    def complete_layer_upload(
+        self,
+        repository_name: str,
+        upload_id: str,
+        layer_digests: list[str],
+        registry_id: str | None = None,
+    ) -> dict[str, Any]:
+        repository = self._get_repository(repository_name, registry_id)
+        upload = repository.uploads.get(upload_id)
+        if upload is None:
+            raise UploadNotFoundException(
+                upload_id, repository_name, repository.registry_id
+            )
+        # Amazon ECR computes the layer digest from the uploaded bytes. Moto does
+        # the same so that BatchCheckLayerAvailability can later find the layer.
+        layer_digest = "sha256:" + hashlib.sha256(bytes(upload.layer_parts)).hexdigest()
+        repository.layers[layer_digest] = len(upload.layer_parts)
+        del repository.uploads[upload_id]
+        return {
+            "registryId": repository.registry_id,
+            "repositoryName": repository_name,
+            "uploadId": upload_id,
+            "layerDigest": layer_digest,
+        }
+
+    def batch_check_layer_availability(
+        self,
+        repository_name: str,
+        layer_digests: list[str],
+        registry_id: str | None = None,
+    ) -> dict[str, Any]:
+        repository = self._get_repository(repository_name, registry_id)
+        layers: list[dict[str, Any]] = []
+        failures: list[dict[str, Any]] = []
+        for digest in layer_digests:
+            if not re.fullmatch(r"sha256:[a-f0-9]{64}", digest):
+                failures.append(
+                    {
+                        "layerDigest": digest,
+                        "failureCode": "InvalidLayerDigest",
+                        "failureReason": f"Invalid layer digest '{digest}'",
+                    }
+                )
+            elif digest in repository.layers:
+                layers.append(
+                    {
+                        "layerDigest": digest,
+                        "layerAvailability": "AVAILABLE",
+                        "layerSize": repository.layers[digest],
+                        "mediaType": LAYER_MEDIA_TYPE,
+                    }
+                )
+            else:
+                failures.append(
+                    {
+                        "layerDigest": digest,
+                        "failureCode": "MissingLayerDigest",
+                        "failureReason": f"Layer digest '{digest}' could not be found",
+                    }
+                )
+        return {"layers": layers, "failures": failures}
+
     def batch_get_image(
         self,
         repository_name: str,

moto/ecr/responses.py

@@ -1,6 +1,6 @@
 import copy
 import time
-from base64 import b64encode
+from base64 import b64decode, b64encode
 from datetime import datetime
 
 from moto.core.responses import ActionResult, BaseResponse, EmptyResult
@@ -96,10 +96,14 @@ def describe_images(self) -> ActionResult:
             dto_images.append(dto_image)
         return ActionResult({"imageDetails": dto_images})
 
-    def batch_check_layer_availability(self) -> None:
-        raise NotImplementedError(
-            "ECR.batch_check_layer_availability is not yet implemented"
+    def batch_check_layer_availability(self) -> ActionResult:
+        repository_name = self._get_param("repositoryName")
+        registry_id = self._get_param("registryId")
+        layer_digests = self._get_param("layerDigests")
+        result = self.ecr_backend.batch_check_layer_availability(
+            repository_name, layer_digests, registry_id=registry_id
         )
+        return ActionResult(result)
 
     def batch_delete_image(self) -> ActionResult:
         repository_str = self._get_param("repositoryName")
@@ -140,8 +144,15 @@ def batch_get_repository_scanning_configuration(self) -> ActionResult:
             }
         )
 
-    def complete_layer_upload(self) -> None:
-        raise NotImplementedError("ECR.complete_layer_upload is not yet implemented")
+    def complete_layer_upload(self) -> ActionResult:
+        repository_name = self._get_param("repositoryName")
+        registry_id = self._get_param("registryId")
+        upload_id = self._get_param("uploadId")
+        layer_digests = self._get_param("layerDigests")
+        result = self.ecr_backend.complete_layer_upload(
+            repository_name, upload_id, layer_digests, registry_id=registry_id
+        )
+        return ActionResult(result)
 
     def delete_repository_policy(self) -> ActionResult:
         registry_id = self._get_param("registryId")
@@ -185,8 +196,13 @@ def get_repository_policy(self) -> ActionResult:
             )
         )
 
-    def initiate_layer_upload(self) -> None:
-        raise NotImplementedError("ECR.initiate_layer_upload is not yet implemented")
+    def initiate_layer_upload(self) -> ActionResult:
+        repository_name = self._get_param("repositoryName")
+        registry_id = self._get_param("registryId")
+        result = self.ecr_backend.initiate_layer_upload(
+            repository_name, registry_id=registry_id
+        )
+        return ActionResult(result)
 
     def set_repository_policy(self) -> ActionResult:
         registry_id = self._get_param("registryId")
@@ -204,8 +220,25 @@ def set_repository_policy(self) -> ActionResult:
             )
         )
 
-    def upload_layer_part(self) -> None:
-        raise NotImplementedError("ECR.upload_layer_part is not yet implemented")
+    def upload_layer_part(self) -> ActionResult:
+        repository_name = self._get_param("repositoryName")
+        registry_id = self._get_param("registryId")
+        upload_id = self._get_param("uploadId")
+        part_first_byte = self._get_param("partFirstByte")
+        part_last_byte = self._get_param("partLastByte")
+        layer_part_blob = self._get_param("layerPartBlob")
+        # The JSON protocol delivers the blob as a base64-encoded string.
+        if isinstance(layer_part_blob, str):
+            layer_part_blob = b64decode(layer_part_blob)
+        result = self.ecr_backend.upload_layer_part(
+            repository_name,
+            upload_id,
+            part_first_byte,
+            part_last_byte,
+            layer_part_blob,
+            registry_id=registry_id,
+        )
+        return ActionResult(result)
 
     def list_tags_for_resource(self) -> ActionResult:
         arn = self._get_param("resourceArn")