Add get_raw_blob

This is needed for my "registry frontend for containers-storage"
project, but should also be useful in general as we've discussed
moving checksum verification into callers. Then there's no
need for a driver etc.

Signed-off-by: Colin Walters <[email protected]>

Author: cgwalters

examples/client.rs

@@ -28,13 +28,24 @@ struct GetBlobOpts {
     size: u64,
 }
 
+#[derive(clap::Parser, Debug)]
+struct FetchContainerToDevNullOpts {
+    #[clap(flatten)]
+    metaopts: GetMetadataOpts,
+
+    /// Use the "raw" path for fetching blobs
+    #[clap(long)]
+    raw_blobs: bool,
+}
+
 /// Simple program to greet a person
 #[derive(clap::Parser, Debug)]
 #[command(version, about, long_about = None)]
 enum Opt {
     GetMetadata(GetMetadataOpts),
     GetBlob(GetBlobOpts),
-    FetchContainerToDevNull(GetMetadataOpts),
+    GetBlobRaw(GetBlobOpts),
+    FetchContainerToDevNull(FetchContainerToDevNullOpts),
 }
 
 #[derive(serde::Serialize, Debug)]
@@ -86,18 +97,49 @@ async fn get_blob(o: GetBlobOpts) -> Result<()> {
     Ok(())
 }
 
-async fn fetch_container_to_devnull(o: GetMetadataOpts) -> Result<()> {
-    let config = o.proxy_opts();
+async fn get_blob_raw(o: GetBlobOpts) -> Result<()> {
+    let proxy = containers_image_proxy::ImageProxy::new().await?;
+    let img = proxy.open_image(&o.reference).await?;
+    let (_, mut datafd, err) = proxy.get_raw_blob(&img, &o.digest).await?;
+
+    let mut stdout = std::io::stdout().lock();
+    let reader = async move {
+        let mut buffer = [0u8; 8192];
+        loop {
+            let n = datafd.read(&mut buffer).await?;
+            if n == 0 {
+                return anyhow::Ok(());
+            }
+            stdout.write_all(&buffer[..n])?;
+        }
+    };
+
+    let (a, b) = tokio::join!(reader, err);
+    a?;
+    b?;
+    Ok(())
+}
+
+async fn fetch_container_to_devnull(o: FetchContainerToDevNullOpts) -> Result<()> {
+    let config = o.metaopts.proxy_opts();
     let proxy = containers_image_proxy::ImageProxy::new_with_config(config).await?;
-    let img = &proxy.open_image(&o.reference).await?;
+    let img = &proxy.open_image(&o.metaopts.reference).await?;
     let manifest = proxy.fetch_manifest(img).await?.1;
     for layer in manifest.layers() {
-        let (mut blob, driver) = proxy.get_descriptor(img, layer).await?;
         let mut devnull = tokio::io::sink();
-        let copier = tokio::io::copy(&mut blob, &mut devnull);
-        let (copier, driver) = tokio::join!(copier, driver);
-        copier?;
-        driver?;
+        if o.raw_blobs {
+            let (_, mut blob, err) = proxy.get_raw_blob(img, layer.digest()).await?;
+            let copier = tokio::io::copy(&mut blob, &mut devnull);
+            let (copier, err) = tokio::join!(copier, err);
+            copier?;
+            err?;
+        } else {
+            let (mut blob, driver) = proxy.get_descriptor(img, layer).await?;
+            let copier = tokio::io::copy(&mut blob, &mut devnull);
+            let (copier, driver) = tokio::join!(copier, driver);
+            copier?;
+            driver?;
+        }
     }
     Ok(())
 }
@@ -106,6 +148,7 @@ async fn run() -> Result<()> {
     match Opt::parse() {
         Opt::GetMetadata(o) => get_metadata(o).await,
         Opt::GetBlob(o) => get_blob(o).await,
+        Opt::GetBlobRaw(o) => get_blob_raw(o).await,
         Opt::FetchContainerToDevNull(o) => fetch_container_to_devnull(o).await,
     }
 }

src/imageproxy.rs

@@ -6,10 +6,11 @@
 
 use cap_std_ext::prelude::CapStdExtCommandExt;
 use cap_std_ext::{cap_std, cap_tempfile};
-use futures_util::Future;
+use futures_util::{Future, FutureExt};
 use oci_spec::image::{Descriptor, Digest};
 use serde::{Deserialize, Serialize};
 use std::fs::File;
+use std::num::NonZeroU32;
 use std::ops::Range;
 use std::os::fd::OwnedFd;
 use std::os::unix::prelude::CommandExt;
@@ -64,6 +65,18 @@ impl Error {
     }
 }
 
+/// Errors returned by get_raw_blob
+#[derive(Error, Debug)]
+#[non_exhaustive]
+pub enum GetBlobError {
+    /// A client may reasonably retry on this type of error.
+    #[error("retryable error")]
+    Retryable(Box<str>),
+    #[error("error")]
+    /// An unknown other error
+    Other(Box<str>),
+}
+
 impl From<rustix::io::Errno> for Error {
     fn from(value: rustix::io::Errno) -> Self {
         Self::Io(value.into())
@@ -318,6 +331,12 @@ pub struct ConvertedLayerInfo {
     pub media_type: oci_spec::image::MediaType,
 }
 
+/// Maps the two types of return values from the proxy
+enum FileDescriptors {
+    FinishPipe { pipeid: NonZeroU32, datafd: OwnedFd },
+    DualFds { datafd: OwnedFd, errfd: OwnedFd },
+}
+
 impl ImageProxy {
     /// Create an image proxy that fetches the target image, using default configuration.
     pub async fn new() -> Result<Self> {
@@ -373,7 +392,7 @@ impl ImageProxy {
     async fn impl_request_raw<T: serde::de::DeserializeOwned + Send + 'static>(
         sockfd: Arc<Mutex<OwnedFd>>,
         req: Request,
-    ) -> Result<(T, Option<(OwnedFd, u32)>)> {
+    ) -> Result<(T, Option<FileDescriptors>)> {
         tracing::trace!("sending request {}", req.method.as_str());
         // TODO: Investigate https://crates.io/crates/uds for SOCK_SEQPACKET tokio
         let r = tokio::task::spawn_blocking(move || {
@@ -394,14 +413,14 @@ impl ImageProxy {
                 rustix::net::RecvFlags::CMSG_CLOEXEC,
             )?
             .bytes;
-            let fdret = cmsg_buffer
+            let mut fdret = cmsg_buffer
                 .drain()
                 .filter_map(|m| match m {
                     rustix::net::RecvAncillaryMessage::ScmRights(f) => Some(f),
                     _ => None,
                 })
                 .flatten()
-                .next();
+                .fuse();
             let buf = &buf[..nread];
             let reply: Reply = serde_json::from_slice(buf)?;
             if !reply.success {
@@ -410,21 +429,42 @@ impl ImageProxy {
                     error: reply.error.into(),
                 });
             }
-            let fdret = match (fdret, reply.pipeid) {
-                (Some(fd), n) => {
-                    if n == 0 {
-                        return Err(Error::Other("got fd but no pipeid".into()));
-                    }
-                    Some((fd, n))
+            let first_fd = fdret.next();
+            let second_fd = fdret.next();
+            if fdret.next().is_some() {
+                return Err(Error::Other(
+                    format!("got more than two file descriptors").into(),
+                ));
+            }
+            let pipeid = NonZeroU32::new(reply.pipeid);
+            let fdret = match (first_fd, second_fd, pipeid) {
+                // No fds, no pipeid
+                (None, None, None) => None,
+                // A FinishPipe instance
+                (Some(datafd), None, Some(pipeid)) => {
+                    Some(FileDescriptors::FinishPipe { pipeid, datafd })
+                }
+                // A dualfd instance
+                (Some(datafd), Some(errfd), None) => {
+                    Some(FileDescriptors::DualFds { datafd, errfd })
+                }
+                // Everything after here is error cases
+                (Some(_), None, None) => {
+                    return Err(Error::Other(format!("got fd with zero pipeid").into()));
+                }
+                (None, Some(_), _) => {
+                    return Err(Error::Other(format!("got errfd with no datafd").into()));
+                }
+                (Some(_), Some(_), Some(n)) => {
+                    return Err(Error::Other(
+                        format!("got pipeid {} with both datafd and errfd", n).into(),
+                    ));
                 }
-                (None, n) => {
-                    if n != 0 {
-                        return Err(Error::Other(format!("got no fd with pipeid {}", n).into()));
-                    }
-                    None
+                (None, _, Some(n)) => {
+                    return Err(Error::Other(format!("got no fd with pipeid {n}").into()));
                 }
             };
-            let reply = serde_json::from_value(reply.value)?;
+            let reply: T = serde_json::from_value(reply.value)?;
             Ok((reply, fdret))
         })
         .await
@@ -438,7 +478,7 @@ impl ImageProxy {
         &self,
         method: &str,
         args: T,
-    ) -> Result<(R, Option<(OwnedFd, u32)>)>
+    ) -> Result<(R, Option<FileDescriptors>)>
     where
         T: IntoIterator<Item = I>,
         I: Into<serde_json::Value>,
@@ -456,9 +496,9 @@ impl ImageProxy {
     }
 
     #[instrument]
-    async fn finish_pipe(&self, pipeid: u32) -> Result<()> {
+    async fn finish_pipe(&self, pipeid: NonZeroU32) -> Result<()> {
         tracing::debug!("closing pipe");
-        let (r, fd) = self.impl_request("FinishPipe", [pipeid]).await?;
+        let (r, fd) = self.impl_request("FinishPipe", [pipeid.get()]).await?;
         if fd.is_some() {
             return Err(Error::Other("Unexpected fd in finish_pipe reply".into()));
         }
@@ -494,9 +534,12 @@ impl ImageProxy {
         Ok(r)
     }
 
-    async fn read_all_fd(&self, fd: Option<(OwnedFd, u32)>) -> Result<Vec<u8>> {
-        let (fd, pipeid) = fd.ok_or_else(|| Error::Other("Missing fd from reply".into()))?;
-        let fd = tokio::fs::File::from_std(std::fs::File::from(fd));
+    async fn read_all_fd(&self, fd: Option<FileDescriptors>) -> Result<Vec<u8>> {
+        let fd = fd.ok_or_else(|| Error::Other("Missing fd from reply".into()))?;
+        let FileDescriptors::FinishPipe { pipeid, datafd } = fd else {
+            return Err(Error::Other("got dualfds, expecting FinishPipe fd".into()));
+        };
+        let fd = tokio::fs::File::from_std(std::fs::File::from(datafd));
         let mut fd = tokio::io::BufReader::new(fd);
         let mut r = Vec::new();
         let reader = fd.read_to_end(&mut r);
@@ -569,13 +612,67 @@ impl ImageProxy {
         let args: Vec<serde_json::Value> =
             vec![img.0.into(), digest.to_string().into(), size.into()];
         let (_bloblen, fd) = self.impl_request::<i64, _, _>("GetBlob", args).await?;
-        let (fd, pipeid) = fd.ok_or_else(|| Error::new_other("Missing fd from reply"))?;
-        let fd = tokio::fs::File::from_std(std::fs::File::from(fd));
+        let fd = fd.ok_or_else(|| Error::Other("Missing fd from reply".into()))?;
+        let FileDescriptors::FinishPipe { pipeid, datafd } = fd else {
+            return Err(Error::Other("got dualfds, expecting FinishPipe fd".into()));
+        };
+        let fd = tokio::fs::File::from_std(std::fs::File::from(datafd));
         let fd = tokio::io::BufReader::new(fd);
         let finish = Box::pin(self.finish_pipe(pipeid));
         Ok((fd, finish))
     }
 
+    async fn read_blob_error(fd: OwnedFd) -> std::result::Result<(), GetBlobError> {
+        let fd = tokio::fs::File::from_std(std::fs::File::from(fd));
+        let mut errfd = tokio::io::BufReader::new(fd);
+        let mut buf = Vec::new();
+        errfd
+            .read_to_end(&mut buf)
+            .await
+            .map_err(|e| GetBlobError::Other(e.to_string().into_boxed_str()))?;
+        if buf.is_empty() {
+            return Ok(());
+        }
+        #[derive(Deserialize)]
+        struct RemoteError {
+            code: String,
+            message: String,
+        }
+        let e: RemoteError = serde_json::from_slice(&buf)
+            .map_err(|e| GetBlobError::Other(e.to_string().into_boxed_str()))?;
+        match e.code.as_str() {
+            // Actually this is OK
+            "EPIPE" => Ok(()),
+            "retryable" => Err(GetBlobError::Retryable(e.message.into_boxed_str())),
+            _ => Err(GetBlobError::Other(e.message.into_boxed_str())),
+        }
+    }
+
+    /// Fetch a blob identified by e.g. `sha256:<digest>`; does not perform
+    /// any verification that the blob matches the digest. The size of the
+    /// blob and a pipe file descriptor are returned.
+    #[instrument]
+    pub async fn get_raw_blob(
+        &self,
+        img: &OpenedImage,
+        digest: &Digest,
+    ) -> Result<(
+        u64,
+        tokio::fs::File,
+        impl Future<Output = std::result::Result<(), GetBlobError>> + Unpin + '_,
+    )> {
+        tracing::debug!("fetching blob");
+        let args: Vec<serde_json::Value> = vec![img.0.into(), digest.to_string().into()];
+        let (bloblen, fd) = self.impl_request::<u64, _, _>("GetRawBlob", args).await?;
+        let fd = fd.ok_or_else(|| Error::new_other("Missing fd from reply"))?;
+        let FileDescriptors::DualFds { datafd, errfd } = fd else {
+            return Err(Error::Other("got single fd, expecting dual fds".into()));
+        };
+        let fd = tokio::fs::File::from_std(std::fs::File::from(datafd));
+        let err = Self::read_blob_error(errfd).boxed();
+        Ok((bloblen, fd, err))
+    }
+
     /// Fetch a descriptor. The requested size and digest are verified (by the proxy process).
     #[instrument]
     pub async fn get_descriptor(