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

@@ -12,9 +12,11 @@ use std::{cmp::Reverse, process::Command, thread::available_parallelism};
 
 use std::{iter::zip, sync::Arc};
 
-use anyhow::{bail, Context, Result};
+use anyhow::{Context, Result, bail};
 use async_compression::tokio::bufread::{GzipDecoder, ZstdDecoder};
-use containers_image_proxy::{ImageProxy, ImageProxyConfig, OpenedImage};
+use containers_image_proxy::{
+    ConvertedLayerInfo, ImageProxy, ImageProxyConfig, OpenedImage, Transport,
+};
 use indicatif::{MultiProgress, ProgressBar, ProgressStyle};
 use oci_spec::image::{Descriptor, ImageConfiguration, ImageManifest, MediaType};
 use rustix::process::geteuid;
@@ -25,7 +27,7 @@ use tokio::{
 
 use composefs::{fsverity::FsVerityHashValue, repository::Repository};
 
-use crate::{config_identifier, layer_identifier, tar::split_async, ContentAndVerity};
+use crate::{ContentAndVerity, config_identifier, layer_identifier, tar::split_async};
 
 // Content type identifiers stored as ASCII in the splitstream file
 pub(crate) const TAR_LAYER_CONTENT_TYPE: u64 = u64::from_le_bytes(*b"ocilayer");
@@ -36,6 +38,7 @@ struct ImageOp<ObjectID: FsVerityHashValue> {
     proxy: ImageProxy,
     img: OpenedImage,
     progress: MultiProgress,
+    transport: Transport,
 }
 
 impl<ObjectID: FsVerityHashValue> ImageOp<ObjectID> {
@@ -44,8 +47,11 @@ impl<ObjectID: FsVerityHashValue> ImageOp<ObjectID> {
         imgref: &str,
         img_proxy_config: Option<ImageProxyConfig>,
     ) -> Result<Self> {
+        // Detect transport from image reference
+        let transport = Transport::try_from(imgref).context("Failed to get image transport")?;
+
         // See https://github.com/containers/skopeo/issues/2563
-        let skopeo_cmd = if imgref.starts_with("containers-storage:") && !geteuid().is_root() {
+        let skopeo_cmd = if transport == Transport::ContainerStorage && !geteuid().is_root() {
             let mut cmd = Command::new("podman");
             cmd.args(["unshare", "skopeo"]);
             Some(cmd)
@@ -86,10 +92,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,46 +114,62 @@ 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), layer_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(layer_size);
 
-            let bar = self.progress.add(ProgressBar::new(descriptor.size()));
+            let bar = self.progress.add(ProgressBar::new(layer_size));
             bar.set_style(ProgressStyle::with_template("[eta {eta}] {bar:40.cyan/blue} {decimal_bytes:>7}/{decimal_total_bytes:7} {msg}")
                 .unwrap()
                 .progress_chars("##-"));
             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 => {
-                    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:?}"),
+            let reader: Box<dyn tokio::io::AsyncBufRead + Unpin + Send> = match self.transport {
+                Transport::ContainerStorage => Box::new(BufReader::new(progress)),
+
+                _ => match descriptor.media_type() {
+                    MediaType::ImageLayer => Box::new(BufReader::new(progress)),
+                    MediaType::ImageLayerGzip => {
+                        Box::new(BufReader::new(GzipDecoder::new(BufReader::new(progress))))
+                    }
+                    MediaType::ImageLayerZstd => {
+                        Box::new(BufReader::new(ZstdDecoder::new(BufReader::new(progress))))
+                    }
+                    other => bail!("Unsupported layer media type {other:?}"),
+                },
             };
 
+            let object_id = split_async(reader, self.repo.clone(), TAR_LAYER_CONTENT_TYPE).await?;
+
             // skopeo is doing data checksums for us to make sure the content we received is equal
             // to the claimed diff_id. We trust it, but we need to check it by awaiting the driver.
             driver.await?;
@@ -194,14 +223,32 @@ 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();
+
+                let layer_idx = manifest_layers
+                    .iter()
+                    .position(|d| *d == descriptor)
+                    .ok_or_else(|| anyhow::anyhow!("Layer descriptor not found in manifest"))?;
+
+                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));
             }