skopeo: Get uncompressed layer when pulling from containers-storage

When pulling from containers storage we were requesting a layer using
the compressed sha digest for that particular layer which caused skopeo
to throw the error

```
failed to invoke method GetBlob: locating item named "sha256:<snip>" for
image with ID "<snip>" (consider removing the image to resolve the issue):
file does not exist
```

To remedy this, when pulling from containers storage we now query using
the `get_blob` method using the uncompressed sha digest for the layer

Related: bootc-dev/bootc#1703

Signed-off-by: Pragyan Poudyal <pragyanpoudyal41999@gmail.com>

Author: Johan-Liebert1

crates/composefs-oci/src/skopeo.rs

@@ -14,7 +14,7 @@ use std::{iter::zip, sync::Arc};
 
 use anyhow::{bail, Context, Result};
 use async_compression::tokio::bufread::{GzipDecoder, ZstdDecoder};
-use containers_image_proxy::{ImageProxy, ImageProxyConfig, OpenedImage};
+use containers_image_proxy::{ConvertedLayerInfo, ImageProxy, ImageProxyConfig, OpenedImage};
 use indicatif::{MultiProgress, ProgressBar, ProgressStyle};
 use oci_spec::image::{Descriptor, ImageConfiguration, ImageManifest, MediaType};
 use rustix::process::geteuid;
@@ -31,11 +31,31 @@ use crate::{config_identifier, layer_identifier, tar::split_async, ContentAndVer
 pub(crate) const TAR_LAYER_CONTENT_TYPE: u64 = u64::from_le_bytes(*b"ocilayer");
 pub(crate) const OCI_CONFIG_CONTENT_TYPE: u64 = u64::from_le_bytes(*b"ociconfg");
 
+/// A backend/transport for OCI/Docker images.
+#[derive(Debug)]
+pub enum Transport {
+    /// A remote Docker/OCI registry (`registry:` or `docker://`)
+    Registry,
+    /// A local OCI directory (`oci:`)
+    OciDir,
+    /// A local OCI archive tarball (`oci-archive:`)
+    OciArchive,
+    /// A local Docker archive tarball (`docker-archive:`)
+    DockerArchive,
+    /// Local container storage (`containers-storage:`)
+    ContainerStorage,
+    /// Local directory (`dir:`)
+    Dir,
+    /// Local Docker daemon (`docker-daemon:`)
+    DockerDaemon,
+}
+
 struct ImageOp<ObjectID: FsVerityHashValue> {
     repo: Arc<Repository<ObjectID>>,
     proxy: ImageProxy,
     img: OpenedImage,
     progress: MultiProgress,
+    transport: Transport,
 }
 
 impl<ObjectID: FsVerityHashValue> ImageOp<ObjectID> {
@@ -44,6 +64,24 @@ impl<ObjectID: FsVerityHashValue> ImageOp<ObjectID> {
         imgref: &str,
         img_proxy_config: Option<ImageProxyConfig>,
     ) -> Result<Self> {
+        // Detect transport from image reference
+        let transport = if imgref.starts_with("containers-storage:") {
+            Transport::ContainerStorage
+        } else if imgref.starts_with("oci:") {
+            Transport::OciDir
+        } else if imgref.starts_with("oci-archive:") {
+            Transport::OciArchive
+        } else if imgref.starts_with("docker-archive:") {
+            Transport::DockerArchive
+        } else if imgref.starts_with("dir:") {
+            Transport::Dir
+        } else if imgref.starts_with("docker-daemon:") {
+            Transport::DockerDaemon
+        } else {
+            // Default to registry for docker:// or registry: prefixes, or bare image names
+            Transport::Registry
+        };
+
         // See https://github.com/containers/skopeo/issues/2563
         let skopeo_cmd = if imgref.starts_with("containers-storage:") && !geteuid().is_root() {
             let mut cmd = Command::new("podman");
@@ -86,10 +124,17 @@ impl<ObjectID: FsVerityHashValue> ImageOp<ObjectID> {
             proxy,
             img,
             progress,
+            transport,
         })
     }
 
-    pub async fn ensure_layer(&self, diff_id: &str, descriptor: &Descriptor) -> Result<ObjectID> {
+    pub async fn ensure_layer(
+        &self,
+        diff_id: &str,
+        descriptor: &Descriptor,
+        uncompressed_layer_info: Arc<Option<Vec<ConvertedLayerInfo>>>,
+        layer_idx: usize,
+    ) -> Result<ObjectID> {
         // We need to use the per_manifest descriptor to download the compressed layer but it gets
         // stored in the repository via the per_config descriptor.  Our return value is the
         // fsverity digest for the corresponding splitstream.
@@ -101,10 +146,37 @@ impl<ObjectID: FsVerityHashValue> ImageOp<ObjectID> {
             Ok(layer_id)
         } else {
             // Otherwise, we need to fetch it...
-            let (blob_reader, driver) = self.proxy.get_descriptor(&self.img, descriptor).await?;
+            let ((blob_reader, driver), size) = match self.transport {
+                Transport::ContainerStorage => {
+                    let layer = uncompressed_layer_info
+                        .as_ref()
+                        .as_ref()
+                        .ok_or_else(|| anyhow::anyhow!("Failed to get uncompressed layer info"))?;
+
+                    let layer = layer.get(layer_idx).ok_or_else(|| {
+                        anyhow::anyhow!(
+                            "Failed to get uncompressed layer info for layer index {layer_idx}"
+                        )
+                    })?;
+
+                    (
+                        self.proxy
+                            .get_blob(&self.img, &layer.digest, layer.size)
+                            .await?,
+                        layer.size,
+                    )
+                }
+
+                _ => (
+                    self.proxy
+                        .get_blob(&self.img, descriptor.digest(), descriptor.size())
+                        .await?,
+                    descriptor.size(),
+                ),
+            };
 
             // See https://github.com/containers/containers-image-proxy-rs/issues/71
-            let blob_reader = blob_reader.take(descriptor.size());
+            let blob_reader = blob_reader.take(size);
 
             let bar = self.progress.add(ProgressBar::new(descriptor.size()));
             bar.set_style(ProgressStyle::with_template("[eta {eta}] {bar:40.cyan/blue} {decimal_bytes:>7}/{decimal_total_bytes:7} {msg}")
@@ -113,32 +185,43 @@ impl<ObjectID: FsVerityHashValue> ImageOp<ObjectID> {
             let progress = bar.wrap_async_read(blob_reader);
             self.progress.println(format!("Fetching layer {diff_id}"))?;
 
-            let object_id = match descriptor.media_type() {
-                MediaType::ImageLayer => {
+            let object_id = match self.transport {
+                Transport::ContainerStorage => {
                     split_async(
                         BufReader::new(progress),
                         self.repo.clone(),
                         TAR_LAYER_CONTENT_TYPE,
                     )
                     .await?
                 }
-                MediaType::ImageLayerGzip => {
-                    split_async(
-                        BufReader::new(GzipDecoder::new(BufReader::new(progress))),
-                        self.repo.clone(),
-                        TAR_LAYER_CONTENT_TYPE,
-                    )
-                    .await?
-                }
-                MediaType::ImageLayerZstd => {
-                    split_async(
-                        BufReader::new(ZstdDecoder::new(BufReader::new(progress))),
-                        self.repo.clone(),
-                        TAR_LAYER_CONTENT_TYPE,
-                    )
-                    .await?
-                }
-                other => bail!("Unsupported layer media type {other:?}"),
+
+                _ => match descriptor.media_type() {
+                    MediaType::ImageLayer => {
+                        split_async(
+                            BufReader::new(progress),
+                            self.repo.clone(),
+                            TAR_LAYER_CONTENT_TYPE,
+                        )
+                        .await?
+                    }
+                    MediaType::ImageLayerGzip => {
+                        split_async(
+                            BufReader::new(GzipDecoder::new(BufReader::new(progress))),
+                            self.repo.clone(),
+                            TAR_LAYER_CONTENT_TYPE,
+                        )
+                        .await?
+                    }
+                    MediaType::ImageLayerZstd => {
+                        split_async(
+                            BufReader::new(ZstdDecoder::new(BufReader::new(progress))),
+                            self.repo.clone(),
+                            TAR_LAYER_CONTENT_TYPE,
+                        )
+                        .await?
+                    }
+                    other => bail!("Unsupported layer media type {other:?}"),
+                },
             };
 
             // skopeo is doing data checksums for us to make sure the content we received is equal
@@ -194,14 +277,33 @@ impl<ObjectID: FsVerityHashValue> ImageOp<ObjectID> {
             let threads = available_parallelism()?;
             let sem = Arc::new(Semaphore::new(threads.into()));
             let mut entries = vec![];
+
+            let uncompressed_layer_info = match self.transport {
+                Transport::ContainerStorage => self.proxy.get_layer_info(&self.img).await?,
+                _ => None,
+            };
+
+            let uncompressed_layer_info = Arc::new(uncompressed_layer_info);
+
             for (mld, diff_id) in layers {
                 let diff_id_ = diff_id.clone();
                 let self_ = Arc::clone(self);
                 let permit = Arc::clone(&sem).acquire_owned().await?;
                 let descriptor = mld.clone();
+
+                // SAFETY: We get descriptor from the manifest_layers itself
+                let layer_idx = manifest_layers
+                    .iter()
+                    .position(|d| *d == descriptor)
+                    .unwrap();
+
+                let uncompressed_layer_info = Arc::clone(&uncompressed_layer_info);
+
                 let future = tokio::spawn(async move {
                     let _permit = permit;
-                    self_.ensure_layer(&diff_id_, &descriptor).await
+                    self_
+                        .ensure_layer(&diff_id_, &descriptor, uncompressed_layer_info, layer_idx)
+                        .await
                 });
                 entries.push((diff_id, future));
             }