hypha: implement custom image fetcher

hypha/Cargo.toml

@@ -17,16 +17,15 @@ walkdir = "2"
 serde = "1.0.195"
 serde_json = "1.0.111"
 sha256 = "1.5.0"
+url = "2.5.0"
+ureq = "2.9.1"
 
 [dependencies.clap]
 version = "4.4.18"
 features = ["derive"]
 
-[dependencies.ocipkg]
-version = "0.2.8"
-
 [dependencies.oci-spec]
-version = "0.5.8"
+version = "0.6.4"
 
 [dependencies.backhand]
 version = "0.14.2"

hypha/src/ctl/mod.rs

@@ -1,7 +1,7 @@
 use crate::error::{HyphaError, Result};
 use crate::image::cache::ImageCache;
+use crate::image::name::ImageName;
 use crate::image::{ImageCompiler, ImageInfo};
-use ocipkg::ImageName;
 use std::fs;
 use std::path::PathBuf;
 use uuid::Uuid;

hypha/src/error.rs

@@ -1,6 +1,8 @@
 use backhand::BackhandError;
+use oci_spec::OciSpecError;
 use std::error::Error;
 use std::fmt::{Display, Formatter};
+use std::num::ParseIntError;
 use std::path::StripPrefixError;
 use xenclient::XenClientError;
 
@@ -43,12 +45,6 @@ impl From<XenClientError> for HyphaError {
     }
 }
 
-impl From<ocipkg::error::Error> for HyphaError {
-    fn from(value: ocipkg::error::Error) -> Self {
-        HyphaError::new(value.to_string().as_str())
-    }
-}
-
 impl From<walkdir::Error> for HyphaError {
     fn from(value: walkdir::Error) -> Self {
         HyphaError::new(value.to_string().as_str())
@@ -72,3 +68,33 @@ impl From<serde_json::Error> for HyphaError {
         HyphaError::new(value.to_string().as_str())
     }
 }
+
+impl From<ureq::Error> for HyphaError {
+    fn from(value: ureq::Error) -> Self {
+        HyphaError::new(value.to_string().as_str())
+    }
+}
+
+impl From<ParseIntError> for HyphaError {
+    fn from(value: ParseIntError) -> Self {
+        HyphaError::new(value.to_string().as_str())
+    }
+}
+
+impl From<OciSpecError> for HyphaError {
+    fn from(value: OciSpecError) -> Self {
+        HyphaError::new(value.to_string().as_str())
+    }
+}
+
+impl From<url::ParseError> for HyphaError {
+    fn from(value: url::ParseError) -> Self {
+        HyphaError::new(value.to_string().as_str())
+    }
+}
+
+impl From<std::fmt::Error> for HyphaError {
+    fn from(value: std::fmt::Error) -> Self {
+        HyphaError::new(value.to_string().as_str())
+    }
+}

hypha/src/image/fetch.rs

@@ -0,0 +1,69 @@
+use crate::error::{HyphaError, Result};
+use oci_spec::image::{Arch, Descriptor, ImageIndex, ImageManifest, MediaType, Os, ToDockerV2S2};
+use std::io::Read;
+use ureq::{Agent, Request, Response};
+use url::Url;
+
+pub struct RegistryClient {
+    agent: Agent,
+    url: Url,
+}
+
+impl RegistryClient {
+    pub fn new(url: Url) -> Result<RegistryClient> {
+        Ok(RegistryClient {
+            agent: Agent::new(),
+            url,
+        })
+    }
+
+    fn call(&mut self, req: Request) -> Result<Response> {
+        Ok(req.call()?)
+    }
+
+    pub fn get_blob(&mut self, name: &str, descriptor: &Descriptor) -> Result<Vec<u8>> {
+        let url = self
+            .url
+            .join(&format!("/v2/{}/blobs/{}", name, descriptor.digest()))?;
+        let response = self.call(self.agent.get(url.as_str()))?;
+        let mut buffer: Vec<u8> = Vec::new();
+        response.into_reader().read_to_end(&mut buffer)?;
+        Ok(buffer)
+    }
+
+    pub fn get_manifest(&mut self, name: &str, reference: &str) -> Result<ImageManifest> {
+        let url = self
+            .url
+            .join(&format!("/v2/{}/manifests/{}", name, reference))?;
+        let accept = format!(
+            "{}, {}, {}",
+            MediaType::ImageManifest.to_docker_v2s2()?,
+            MediaType::ImageManifest,
+            MediaType::ImageIndex,
+        );
+        let response = self.call(self.agent.get(url.as_str()).set("Accept", &accept))?;
+        let content_type = response.header("Content-Type").ok_or_else(|| {
+            HyphaError::new("registry response did not have a Content-Type header")
+        })?;
+        if content_type == MediaType::ImageIndex.to_string() {
+            let index = ImageIndex::from_reader(response.into_reader())?;
+            let descriptor = self
+                .pick_manifest(index)
+                .ok_or_else(|| HyphaError::new("unable to pick manifest from index"))?;
+            return self.get_manifest(name, descriptor.digest());
+        }
+        let manifest = ImageManifest::from_reader(response.into_reader())?;
+        Ok(manifest)
+    }
+
+    fn pick_manifest(&mut self, index: ImageIndex) -> Option<Descriptor> {
+        for item in index.manifests() {
+            if let Some(platform) = item.platform() {
+                if *platform.os() == Os::Linux && *platform.architecture() == Arch::Amd64 {
+                    return Some(item.clone());
+                }
+            }
+        }
+        None
+    }
+}

hypha/src/image/mod.rs

@@ -1,13 +1,14 @@
 pub mod cache;
+pub mod fetch;
+pub mod name;
 
 use crate::error::{HyphaError, Result};
 use crate::image::cache::ImageCache;
+use crate::image::fetch::RegistryClient;
+use crate::image::name::ImageName;
 use backhand::{FilesystemWriter, NodeHeader};
 use log::{debug, trace};
 use oci_spec::image::{ImageConfiguration, ImageManifest, MediaType};
-use ocipkg::distribution::Client;
-use ocipkg::error::Error;
-use ocipkg::{Digest, ImageName};
 use std::fs;
 use std::fs::File;
 use std::io::BufReader;
@@ -71,11 +72,8 @@ impl ImageCompiler<'_> {
             "ImageCompiler download image={image}, image_dir={}",
             image_dir.to_str().unwrap()
         );
-        let ImageName {
-            name, reference, ..
-        } = image;
-        let mut client = Client::new(image.registry_url()?, name.clone())?;
-        let manifest = client.get_manifest(reference)?;
+        let mut client = RegistryClient::new(image.registry_url()?)?;
+        let manifest = client.get_manifest(&image.name, &image.reference)?;
         let manifest_serialized = serde_json::to_string(&manifest)?;
         let cache_key = format!(
             "manifest\n{}squashfs-version\n{}\n",
@@ -87,7 +85,7 @@ impl ImageCompiler<'_> {
             return Ok(cached);
         }
 
-        let config_bytes = client.get_blob(&Digest::new(manifest.config().digest())?)?;
+        let config_bytes = client.get_blob(&image.name, manifest.config())?;
         let config: ImageConfiguration = serde_json::from_slice(&config_bytes)?;
 
         for layer in manifest.layers() {
@@ -97,7 +95,7 @@ impl ImageCompiler<'_> {
                 layer.size()
             );
 
-            let blob = client.get_blob(&Digest::new(layer.digest())?)?;
+            let blob = client.get_blob(&image.name, layer)?;
             match layer.media_type() {
                 MediaType::ImageLayerGzip => {}
                 MediaType::Other(ty) => {
@@ -123,7 +121,7 @@ impl ImageCompiler<'_> {
             let info = ImageInfo::new(squash_file.clone(), manifest.clone(), config)?;
             return self.cache.store(&cache_digest, &info);
         }
-        Err(Error::MissingLayer.into())
+        Err(HyphaError::new("unable to find image layer"))
     }
 
     fn squash(&self, image_dir: &PathBuf, squash_file: &PathBuf) -> Result<()> {

hypha/src/image/name.rs

@@ -0,0 +1,67 @@
+use crate::error::Result;
+use std::fmt;
+use url::Url;
+
+#[derive(Debug, Clone, PartialEq, Eq, Hash)]
+pub struct ImageName {
+    pub hostname: String,
+    pub port: Option<u16>,
+    pub name: String,
+    pub reference: String,
+}
+
+impl fmt::Display for ImageName {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        if let Some(port) = self.port {
+            write!(
+                f,
+                "{}:{}/{}:{}",
+                self.hostname, port, self.name, self.reference
+            )
+        } else {
+            write!(f, "{}/{}:{}", self.hostname, self.name, self.reference)
+        }
+    }
+}
+
+impl Default for ImageName {
+    fn default() -> Self {
+        Self::parse(&format!("{}", uuid::Uuid::new_v4().as_hyphenated()))
+            .expect("UUID hyphenated must be valid name")
+    }
+}
+
+impl ImageName {
+    pub fn parse(name: &str) -> Result<Self> {
+        let (hostname, name) = name
+            .split_once('/')
+            .unwrap_or(("registry-1.docker.io", name));
+        let (hostname, port) = if let Some((hostname, port)) = hostname.split_once(':') {
+            (hostname, Some(str::parse(port)?))
+        } else {
+            (hostname, None)
+        };
+        let (name, reference) = name.split_once(':').unwrap_or((name, "latest"));
+        Ok(ImageName {
+            hostname: hostname.to_string(),
+            port,
+            name: name.to_string(),
+            reference: reference.to_string(),
+        })
+    }
+
+    /// URL for OCI distribution API endpoint
+    pub fn registry_url(&self) -> Result<Url> {
+        let hostname = if let Some(port) = self.port {
+            format!("{}:{}", self.hostname, port)
+        } else {
+            self.hostname.clone()
+        };
+        let url = if self.hostname.starts_with("localhost") {
+            format!("http://{}", hostname)
+        } else {
+            format!("https://{}", hostname)
+        };
+        Ok(Url::parse(&url)?)
+    }
+}