From 08ad0701819267355df10b3904e601523096554e Mon Sep 17 00:00:00 2001 From: Nikolay Govorov Date: Thu, 16 Jul 2026 01:00:47 +0100 Subject: Async typed hule-oci --- Cargo.lock | 1 + crates/hule-oci/Cargo.toml | 1 + crates/hule-oci/src/error.rs | 119 ++++++ crates/hule-oci/src/lib.rs | 715 +++++++++++++++++---------------- crates/hule-oci/src/storage.rs | 213 ++++++++++ crates/hule/src/main.rs | 45 ++- 6 files changed, 739 insertions(+), 355 deletions(-) create mode 100644 crates/hule-oci/src/error.rs create mode 100644 crates/hule-oci/src/storage.rs diff --git a/Cargo.lock b/Cargo.lock index cff42e0..d2d38d6 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -690,6 +690,7 @@ dependencies = [ "oci-client", "serde_json", "sha2 0.10.9", + "tokio", "zstd", ] diff --git a/crates/hule-oci/Cargo.toml b/crates/hule-oci/Cargo.toml index 9fa92a9..c395abd 100644 --- a/crates/hule-oci/Cargo.toml +++ b/crates/hule-oci/Cargo.toml @@ -15,6 +15,7 @@ repository.workspace = true [dependencies] hule-hmi.workspace = true serde_json.workspace = true +tokio = { workspace = true, features = ["fs", "io-util", "rt"] } oci-client = "0.17" zstd = "0.13" sha2 = "0.10" diff --git a/crates/hule-oci/src/error.rs b/crates/hule-oci/src/error.rs new file mode 100644 index 0000000..0a5542c --- /dev/null +++ b/crates/hule-oci/src/error.rs @@ -0,0 +1,119 @@ +// SPDX-FileCopyrightText: 2026 Nikolay Govorov +// SPDX-License-Identifier: Apache-2.0 + +use oci_client::errors::OciDistributionError; +use std::fmt; +use std::path::PathBuf; + +#[derive(Debug)] +pub enum Error { + Io(std::io::Error), + Json(serde_json::Error), + InvalidImage(hule_hmi::ParseError), + InvalidReference { + reference: String, + source: oci_client::ParseError, + }, + Registry(OciDistributionError), + Task(tokio::task::JoinError), + InvalidImagePath(PathBuf), + MissingConfig(PathBuf), + ImageNotFound(String), + MissingManifest, + InvalidLayer(&'static str), + InvalidAnnotation { + name: &'static str, + value: String, + }, + EmptyImageFile(PathBuf), + InvalidStoragePath { + kind: &'static str, + value: String, + }, +} + +impl fmt::Display for Error { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::Io(error) => error.fmt(f), + Self::Json(error) => error.fmt(f), + Self::InvalidImage(error) => error.fmt(f), + Self::InvalidReference { reference, source } => { + write!(f, "invalid reference '{reference}': {source}") + } + Self::Registry(error) => error.fmt(f), + Self::Task(error) => write!(f, "background task failed: {error}"), + Self::InvalidImagePath(path) => { + write!(f, "{} has no file name", path.display()) + } + Self::MissingConfig(path) => { + write!(f, "{} does not contain config.json", path.display()) + } + Self::ImageNotFound(reference) => { + write!(f, "no local image tagged '{reference}'") + } + Self::MissingManifest => f.write_str("registry returned no image manifest"), + Self::InvalidLayer(reason) => write!(f, "invalid chunk layer: {reason}"), + Self::InvalidAnnotation { name, value } => { + write!(f, "invalid chunk {name} annotation '{value}'") + } + Self::EmptyImageFile(path) => write!(f, "{} is empty", path.display()), + Self::InvalidStoragePath { kind, value } => { + write!(f, "invalid {kind} '{value}'") + } + } + } +} + +impl std::error::Error for Error { + fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { + match self { + Self::Io(error) => Some(error), + Self::Json(error) => Some(error), + Self::InvalidImage(error) => Some(error), + Self::InvalidReference { source, .. } => Some(source), + Self::Registry(error) => Some(error), + Self::Task(error) => Some(error), + Self::InvalidImagePath(_) + | Self::MissingConfig(_) + | Self::ImageNotFound(_) + | Self::MissingManifest + | Self::InvalidLayer(_) + | Self::InvalidAnnotation { .. } + | Self::EmptyImageFile(_) + | Self::InvalidStoragePath { .. } => None, + } + } +} + +impl From for Error { + fn from(error: std::io::Error) -> Self { + Self::Io(error) + } +} + +impl From for Error { + fn from(error: serde_json::Error) -> Self { + Self::Json(error) + } +} + +impl From for Error { + fn from(error: hule_hmi::ParseError) -> Self { + Self::InvalidImage(error) + } +} + +impl From for Error { + fn from(error: OciDistributionError) -> Self { + Self::Registry(error) + } +} + +impl From for Error { + fn from(error: tokio::task::JoinError) -> Self { + Self::Task(error) + } +} + +pub type Result = std::result::Result; diff --git a/crates/hule-oci/src/lib.rs b/crates/hule-oci/src/lib.rs index e53615b..990bd87 100644 --- a/crates/hule-oci/src/lib.rs +++ b/crates/hule-oci/src/lib.rs @@ -3,6 +3,9 @@ //! OCI image storage and registry operations for Hule machine images. +mod error; +mod storage; + use hule_hmi::MachineImage; use oci_client::annotations::{ORG_OPENCONTAINERS_IMAGE_REF_NAME, ORG_OPENCONTAINERS_IMAGE_TITLE}; use oci_client::client::{ClientConfig, ClientProtocol, Config, ImageLayer}; @@ -12,9 +15,14 @@ use oci_client::manifest::{ use oci_client::secrets::RegistryAuth; use oci_client::{Client, Reference}; use std::collections::{BTreeMap, BTreeSet}; -use std::io::Read; use std::path::{Path, PathBuf}; -pub type Result = std::result::Result; +use tokio::fs::{self, File}; +use tokio::io::AsyncReadExt; +use tokio::task; + +pub use error::{Error, Result}; +pub use storage::Storage; + type R = Result; // ---- local OCI-layout store ------------------------------------------- @@ -25,82 +33,6 @@ const HULE_CONFIG_MEDIA_TYPE: &str = "application/vnd.hule.machine.config.v1+jso const ANNOTATION_CHUNK_OFFSET: &str = "io.hule.chunk.offset"; const ANNOTATION_CHUNK_LENGTH: &str = "io.hule.chunk.length"; -fn store_root() -> R { - let home = std::env::var("HOME").map_err(|_| "HOME is not set".to_string())?; - Ok(PathBuf::from(home).join(".hule").join("store")) -} - -fn blobs_dir() -> R { - Ok(store_root()?.join("blobs").join("sha256")) -} - -fn materialized_dir(manifest_digest: &str) -> R { - Ok(store_root()? - .join("materialized") - .join(strip_sha256(manifest_digest))) -} - -fn index_path() -> R { - Ok(store_root()?.join("index.json")) -} - -fn strip_sha256(digest: &str) -> &str { - digest.strip_prefix("sha256:").unwrap_or(digest) -} - -fn sha256_hex(data: &[u8]) -> String { - use sha2::{Digest, Sha256}; - let mut hasher = Sha256::new(); - hasher.update(data); - format!("sha256:{:x}", hasher.finalize()) -} - -fn ensure_store() -> R<()> { - std::fs::create_dir_all(blobs_dir()?).map_err(|e| e.to_string())?; - std::fs::create_dir_all(store_root()?.join("materialized")).map_err(|e| e.to_string())?; - let layout = store_root()?.join("oci-layout"); - if !layout.exists() { - std::fs::write(&layout, br#"{"imageLayoutVersion":"1.0.0"}"#).map_err(|e| e.to_string())?; - } - Ok(()) -} - -/// Writes raw bytes as a content-addressed blob, no-op if already present. -fn write_blob(data: &[u8]) -> R { - let digest = sha256_hex(data); - let path = blobs_dir()?.join(strip_sha256(&digest)); - if !path.exists() { - let tmp = path.with_extension("tmp"); - std::fs::write(&tmp, data).map_err(|e| e.to_string())?; - std::fs::rename(&tmp, &path).map_err(|e| e.to_string())?; - } - Ok(digest) -} - -fn read_blob(digest: &str) -> R> { - std::fs::read(blobs_dir()?.join(strip_sha256(digest))).map_err(|e| e.to_string()) -} - -fn read_index() -> R { - let path = index_path()?; - if !path.exists() { - return Ok(OciImageIndex { - schema_version: 2, - media_type: None, - manifests: vec![], - artifact_type: None, - annotations: None, - }); - } - let data = std::fs::read(&path).map_err(|e| e.to_string())?; - serde_json::from_slice(&data).map_err(|e| e.to_string()) -} - -fn write_index(index: &OciImageIndex) -> R<()> { - let data = serde_json::to_vec_pretty(index).map_err(|e| e.to_string())?; - std::fs::write(index_path()?, data).map_err(|e| e.to_string()) -} - fn index_lookup(index: &OciImageIndex, reference: &str) -> Option { index .manifests @@ -138,79 +70,71 @@ fn index_set(index: &mut OciImageIndex, reference: &str, digest: &str, size: u64 }); } -fn hardlink_or_copy(src: &Path, dest: &Path) -> R<()> { - if std::fs::hard_link(src, dest).is_err() { - std::fs::copy(src, dest).map_err(|e| e.to_string())?; - } - Ok(()) +/// Manages Hule images backed by a [`Storage`]. +#[derive(Clone, Debug)] +pub struct ImageManager { + storage: Storage, } -fn write_at(file: &std::fs::File, offset: u64, data: &[u8]) -> R<()> { - use std::os::unix::fs::FileExt; - let mut written = 0usize; - while written < data.len() { - let n = file - .write_at(&data[written..], offset + written as u64) - .map_err(|e| e.to_string())?; - if n == 0 { - return Err("write_at wrote 0 bytes".to_string()); - } - written += n; +impl ImageManager { + pub fn new(storage: Storage) -> Self { + Self { storage } } - Ok(()) -} -/// Splits `path` into fixed-size chunks, compresses each independently -/// (own zstd frame, no state shared between chunks -- see the design notes -/// in the project plan on why this must not be done the other way around), -/// and writes each as its own blob. Returns one descriptor per chunk, all -/// sharing a title annotation plus a chunk offset/length in terms of the -/// *uncompressed* file so pull can reassemble it. -fn write_file_chunks(path: &Path) -> R> { - let title = path - .file_name() - .ok_or_else(|| format!("{} has no file name", path.display()))? - .to_string_lossy() - .to_string(); - let mut file = std::fs::File::open(path).map_err(|e| e.to_string())?; - let mut descriptors = Vec::new(); - let mut offset: u64 = 0; - loop { - let mut buf = Vec::with_capacity(CHUNK_SIZE); - (&mut file) - .take(CHUNK_SIZE as u64) - .read_to_end(&mut buf) - .map_err(|e| e.to_string())?; - if buf.is_empty() { - break; - } - let length = buf.len() as u64; - let compressed = zstd::stream::encode_all(&buf[..], 0).map_err(|e| e.to_string())?; - let digest = write_blob(&compressed)?; - - let mut annotations = BTreeMap::new(); - annotations.insert(ORG_OPENCONTAINERS_IMAGE_TITLE.to_string(), title.clone()); - annotations.insert(ANNOTATION_CHUNK_OFFSET.to_string(), offset.to_string()); - annotations.insert(ANNOTATION_CHUNK_LENGTH.to_string(), length.to_string()); - - descriptors.push(OciDescriptor { - media_type: HULE_CHUNK_MEDIA_TYPE.to_string(), - digest, - size: compressed.len() as i64, - urls: None, - annotations: Some(annotations), - artifact_type: None, - }); + pub fn storage(&self) -> &Storage { + &self.storage + } - offset += length; - if length < CHUNK_SIZE as u64 { - break; + /// Splits `path` into fixed-size chunks, compresses each independently + /// and writes each as its own blob. + async fn write_file_chunks(&self, path: &Path) -> R> { + let title = path + .file_name() + .ok_or_else(|| Error::InvalidImagePath(path.to_path_buf()))? + .to_string_lossy() + .to_string(); + let mut file = File::open(path).await?; + let mut descriptors = Vec::new(); + let mut offset: u64 = 0; + loop { + let mut buf = Vec::with_capacity(CHUNK_SIZE); + (&mut file) + .take(CHUNK_SIZE as u64) + .read_to_end(&mut buf) + .await?; + if buf.is_empty() { + break; + } + let length = buf.len() as u64; + let compressed = + task::spawn_blocking(move || zstd::stream::encode_all(&buf[..], 0)).await??; + let compressed_len = compressed.len(); + let (digest, _) = self.storage.write_blob(compressed).await?; + + let mut annotations = BTreeMap::new(); + annotations.insert(ORG_OPENCONTAINERS_IMAGE_TITLE.to_string(), title.clone()); + annotations.insert(ANNOTATION_CHUNK_OFFSET.to_string(), offset.to_string()); + annotations.insert(ANNOTATION_CHUNK_LENGTH.to_string(), length.to_string()); + + descriptors.push(OciDescriptor { + media_type: HULE_CHUNK_MEDIA_TYPE.to_string(), + digest, + size: compressed_len as i64, + urls: None, + annotations: Some(annotations), + artifact_type: None, + }); + + offset += length; + if length < CHUNK_SIZE as u64 { + break; + } } + if descriptors.is_empty() { + return Err(Error::EmptyImageFile(path.to_path_buf())); + } + Ok(descriptors) } - if descriptors.is_empty() { - return Err(format!("{} is empty", path.display())); - } - Ok(descriptors) } fn make_client(reference: &Reference) -> Client { @@ -228,232 +152,335 @@ fn make_client(reference: &Reference) -> Client { } fn parse_reference(s: &str) -> R { - s.parse() - .map_err(|e| format!("invalid reference '{s}': {e}")) + s.parse().map_err(|source| Error::InvalidReference { + reference: s.to_string(), + source, + }) } -// ---- import / push / pull / run --------------------------------------- +// ---- image operations ------------------------------------------------- + +impl ImageManager { + /// Loads a prepared HMI image directory and optionally tags it. + pub async fn load(&self, path: &Path, reference: Option<&str>) -> R { + let config_path = path.join("config.json"); + match fs::metadata(&config_path).await { + Ok(metadata) if metadata.is_file() => {} + Ok(_) => return Err(Error::MissingConfig(path.to_path_buf())), + Err(error) if error.kind() == std::io::ErrorKind::NotFound => { + return Err(Error::MissingConfig(path.to_path_buf())); + } + Err(error) => return Err(error.into()), + } + let config_bytes = fs::read(&config_path).await?; + MachineImage::from_json(&config_bytes)?; -/// Imports an image directory into the local OCI layout and optionally tags it. -pub async fn import(path: &Path, reference: Option<&str>) -> R { - ensure_store()?; - let config_path = path.join("config.json"); - if !config_path.is_file() { - return Err(format!("{} does not contain config.json", path.display())); - } - let config_bytes = std::fs::read(&config_path).map_err(|e| e.to_string())?; - MachineImage::from_json(&config_bytes).map_err(|e| e.to_string())?; - - let mut entries: Vec<_> = std::fs::read_dir(path) - .map_err(|e| e.to_string())? - .collect::, _>>() - .map_err(|e| e.to_string())?; - entries.sort_by_key(|e| e.file_name()); - - let mut layers = Vec::new(); - let mut sources: Vec<(String, PathBuf)> = Vec::new(); - for entry in entries { - if !entry.file_type().map_err(|e| e.to_string())?.is_file() { - continue; + let mut dir = fs::read_dir(path).await?; + let mut entries = Vec::new(); + while let Some(entry) = dir.next_entry().await? { + entries.push(entry); } - let name = entry.file_name().to_string_lossy().to_string(); - if name == "config.json" { - continue; + entries.sort_by_key(|entry| entry.file_name()); + + let mut layers = Vec::new(); + let mut sources: Vec<(String, PathBuf)> = Vec::new(); + for entry in entries { + if !entry.file_type().await?.is_file() { + continue; + } + let name = entry.file_name().to_string_lossy().to_string(); + if name == "config.json" { + continue; + } + let file_path = entry.path(); + layers.extend(self.write_file_chunks(&file_path).await?); + sources.push((name, file_path)); } - let file_path = entry.path(); - layers.extend(write_file_chunks(&file_path)?); - sources.push((name, file_path)); - } - let config_digest = write_blob(&config_bytes)?; - let config = OciDescriptor { - media_type: HULE_CONFIG_MEDIA_TYPE.to_string(), - digest: config_digest, - size: config_bytes.len() as i64, - urls: None, - annotations: None, - artifact_type: None, - }; + let config_size = config_bytes.len(); + let (config_digest, _) = self.storage.write_blob(config_bytes).await?; + let config = OciDescriptor { + media_type: HULE_CONFIG_MEDIA_TYPE.to_string(), + digest: config_digest, + size: config_size as i64, + urls: None, + annotations: None, + artifact_type: None, + }; - let manifest = OciImageManifest { - schema_version: 2, - media_type: Some(OCI_IMAGE_MEDIA_TYPE.to_string()), - config, - layers, - subject: None, - artifact_type: None, - annotations: None, - }; - let manifest_bytes = serde_json::to_vec(&manifest).map_err(|e| e.to_string())?; - let manifest_digest = write_blob(&manifest_bytes)?; - - let materialized = materialized_dir(&manifest_digest)?; - std::fs::create_dir_all(&materialized).map_err(|e| e.to_string())?; - for (title, source) in &sources { - let dest = materialized.join(title); - if !dest.exists() { - hardlink_or_copy(source, &dest)?; + let manifest = OciImageManifest { + schema_version: 2, + media_type: Some(OCI_IMAGE_MEDIA_TYPE.to_string()), + config, + layers, + subject: None, + artifact_type: None, + annotations: None, + }; + let manifest_bytes = serde_json::to_vec(&manifest)?; + let manifest_size = manifest_bytes.len(); + let (manifest_digest, _) = self.storage.write_blob(manifest_bytes).await?; + + self.storage.ensure_hmi_image(&manifest_digest).await?; + for (title, source) in &sources { + self.storage + .cache_hmi_file(&manifest_digest, title, source) + .await?; } - } - - if let Some(reference) = reference { - let mut index = read_index()?; - index_set( - &mut index, - reference, - &manifest_digest, - manifest_bytes.len() as u64, - ); - write_index(&index)?; - } - Ok(manifest_digest) -} -/// Pushes a locally tagged image to its registry reference. -pub async fn push(reference_str: &str) -> R<()> { - ensure_store()?; - let index = read_index()?; - let manifest_digest = index_lookup(&index, reference_str) - .ok_or_else(|| format!("no local image tagged '{reference_str}'"))?; - let manifest_bytes = read_blob(&manifest_digest)?; - let manifest: OciImageManifest = - serde_json::from_slice(&manifest_bytes).map_err(|e| e.to_string())?; - - let reference = parse_reference(reference_str)?; - let client = make_client(&reference); - let auth = RegistryAuth::Anonymous; - - let mut layers = Vec::new(); - for descriptor in &manifest.layers { - let data = read_blob(&descriptor.digest)?; - layers.push(ImageLayer::new( - data, - descriptor.media_type.clone(), - descriptor.annotations.clone(), - )); + if let Some(reference) = reference { + let mut index = self.storage.read_index().await?; + index_set( + &mut index, + reference, + &manifest_digest, + manifest_size as u64, + ); + self.storage.write_index(&index).await?; + } + Ok(manifest_digest) } - let config_bytes = read_blob(&manifest.config.digest)?; - let config = Config::new( - config_bytes, - manifest.config.media_type.clone(), - manifest.config.annotations.clone(), - ); - client - .push(&reference, &layers, config, &auth, Some(manifest)) - .await - .map_err(|e| e.to_string())?; + /// Pushes a locally tagged image to its registry reference. + pub async fn push(&self, reference_str: &str) -> R<()> { + let index = self.storage.read_index().await?; + let manifest_digest = index_lookup(&index, reference_str) + .ok_or_else(|| Error::ImageNotFound(reference_str.to_string()))?; + let manifest_bytes = self.storage.read_blob(&manifest_digest).await?; + let manifest: OciImageManifest = serde_json::from_slice(&manifest_bytes)?; + + let reference = parse_reference(reference_str)?; + let client = make_client(&reference); + let auth = RegistryAuth::Anonymous; + + let mut layers = Vec::new(); + for descriptor in &manifest.layers { + let data = self.storage.read_blob(&descriptor.digest).await?; + layers.push(ImageLayer::new( + data, + descriptor.media_type.clone(), + descriptor.annotations.clone(), + )); + } + let config_bytes = self.storage.read_blob(&manifest.config.digest).await?; + let config = Config::new( + config_bytes, + manifest.config.media_type.clone(), + manifest.config.annotations.clone(), + ); - Ok(()) -} + client + .push(&reference, &layers, config, &auth, Some(manifest)) + .await?; -/// Pulls an image into the local OCI layout and materializes its files. -pub async fn pull(reference_str: &str) -> R { - ensure_store()?; - let reference = parse_reference(reference_str)?; - let client = make_client(&reference); - let auth = RegistryAuth::Anonymous; - - let image_data = client - .pull(&reference, &auth, vec![HULE_CHUNK_MEDIA_TYPE]) - .await - .map_err(|e| e.to_string())?; - let manifest = image_data - .manifest - .ok_or("registry returned no image manifest")?; - - // `client.pull()` fetches layers via `buffer_unordered`, so `image_data.layers` is in - // completion order, NOT `manifest.layers` order -- do not zip the two. Each `ImageLayer` - // already carries its own annotations, so it's self-describing without cross-referencing. - for layer in &image_data.layers { - write_blob(&layer.data)?; + Ok(()) } - write_blob(&image_data.config.data)?; - let manifest_bytes = serde_json::to_vec(&manifest).map_err(|e| e.to_string())?; - let manifest_digest = write_blob(&manifest_bytes)?; + /// Pulls an image into the OCI layout and materializes it in the HMI store. + pub async fn pull(&self, reference_str: &str) -> R { + let reference = parse_reference(reference_str)?; + let client = make_client(&reference); + let auth = RegistryAuth::Anonymous; + + let mut image_data = client + .pull(&reference, &auth, vec![HULE_CHUNK_MEDIA_TYPE]) + .await?; + let manifest = image_data.manifest.ok_or(Error::MissingManifest)?; + + // `client.pull()` fetches layers via `buffer_unordered`, so `image_data.layers` is + // in completion order, not manifest order. Each layer carries its own annotations. + for layer in &mut image_data.layers { + let data = std::mem::take(&mut layer.data); + let (_, data) = self.storage.write_blob(data).await?; + layer.data = data; + } + let config_data = std::mem::take(&mut image_data.config.data); + let (_, config_data) = self.storage.write_blob(config_data).await?; + image_data.config.data = config_data; + + let manifest_bytes = serde_json::to_vec(&manifest)?; + let manifest_size = manifest_bytes.len(); + let (manifest_digest, _) = self.storage.write_blob(manifest_bytes).await?; + self.storage.ensure_hmi_image(&manifest_digest).await?; + + let mut by_title: BTreeMap> = BTreeMap::new(); + for layer in image_data.layers { + let annotations = layer + .annotations + .as_ref() + .ok_or(Error::InvalidLayer("missing annotations"))?; + let title = annotations + .get(ORG_OPENCONTAINERS_IMAGE_TITLE) + .ok_or(Error::InvalidLayer("missing title annotation"))?; + let offset_value = annotations + .get(ANNOTATION_CHUNK_OFFSET) + .ok_or(Error::InvalidLayer("missing offset annotation"))?; + let offset: u64 = offset_value.parse().map_err(|_| Error::InvalidAnnotation { + name: "offset", + value: offset_value.clone(), + })?; + by_title + .entry(title.clone()) + .or_default() + .push((offset, layer)); + } - let materialized = materialized_dir(&manifest_digest)?; - std::fs::create_dir_all(&materialized).map_err(|e| e.to_string())?; + for (title, chunks) in by_title { + let mut decompressed_chunks = Vec::with_capacity(chunks.len()); + for (offset, layer) in chunks { + let data = task::spawn_blocking(move || zstd::stream::decode_all(&layer.data[..])) + .await??; + decompressed_chunks.push((offset, data)); + } + self.storage + .write_hmi_file(&manifest_digest, &title, decompressed_chunks) + .await?; + } - let mut by_title: BTreeMap> = BTreeMap::new(); - for layer in &image_data.layers { - let annotations = layer - .annotations - .as_ref() - .ok_or("chunk layer missing annotations")?; - let title = annotations - .get(ORG_OPENCONTAINERS_IMAGE_TITLE) - .ok_or("chunk layer missing title annotation")?; - let offset: u64 = annotations - .get(ANNOTATION_CHUNK_OFFSET) - .ok_or("chunk layer missing offset annotation")? - .parse() - .map_err(|_| "invalid chunk offset annotation".to_string())?; - by_title - .entry(title.clone()) - .or_default() - .push((offset, &layer.data[..])); + let mut index = self.storage.read_index().await?; + index_set( + &mut index, + reference_str, + &manifest_digest, + manifest_size as u64, + ); + self.storage.write_index(&index).await?; + + Ok(manifest_digest) } - for (title, mut chunks) in by_title { - chunks.sort_by_key(|(offset, _)| *offset); - let file = std::fs::File::create(materialized.join(&title)).map_err(|e| e.to_string())?; - for (offset, compressed) in chunks { - let decompressed = zstd::stream::decode_all(compressed).map_err(|e| e.to_string())?; - write_at(&file, offset, &decompressed)?; - } + /// Returns whether a reference is present in the local OCI layout. + pub async fn contains(&self, reference_str: &str) -> R { + Ok(index_lookup(&self.storage.read_index().await?, reference_str).is_some()) } - let mut index = read_index()?; - index_set( - &mut index, - reference_str, - &manifest_digest, - manifest_bytes.len() as u64, - ); - write_index(&index)?; + /// Materializes a locally tagged image into `destination` and returns config.json. + pub async fn materialize(&self, reference_str: &str, destination: &Path) -> R> { + let manifest_digest = index_lookup(&self.storage.read_index().await?, reference_str) + .ok_or_else(|| Error::ImageNotFound(reference_str.to_string()))?; - Ok(manifest_digest) -} + let manifest_bytes = self.storage.read_blob(&manifest_digest).await?; + let manifest: OciImageManifest = serde_json::from_slice(&manifest_bytes)?; + let config_bytes = self.storage.read_blob(&manifest.config.digest).await?; -/// Returns whether a reference is present in the local OCI layout. -pub fn contains(reference_str: &str) -> R { - ensure_store()?; - Ok(index_lookup(&read_index()?, reference_str).is_some()) -} + fs::create_dir_all(destination).await?; + fs::write(destination.join("config.json"), &config_bytes).await?; -/// Materializes a locally tagged image into `destination` and returns config.json. -pub fn materialize(reference_str: &str, destination: &Path) -> R> { - ensure_store()?; - let manifest_digest = index_lookup(&read_index()?, reference_str) - .ok_or_else(|| format!("no local image tagged '{reference_str}'"))?; - - let manifest_bytes = read_blob(&manifest_digest)?; - let manifest: OciImageManifest = - serde_json::from_slice(&manifest_bytes).map_err(|e| e.to_string())?; - let config_bytes = read_blob(&manifest.config.digest)?; - - std::fs::create_dir_all(destination).map_err(|e| e.to_string())?; - std::fs::write(destination.join("config.json"), &config_bytes).map_err(|e| e.to_string())?; - - let materialized = materialized_dir(&manifest_digest)?; - let mut titles: BTreeSet = BTreeSet::new(); - for descriptor in &manifest.layers { - if let Some(title) = descriptor - .annotations - .as_ref() - .and_then(|a| a.get(ORG_OPENCONTAINERS_IMAGE_TITLE)) - { - titles.insert(title.clone()); + let mut titles: BTreeSet = BTreeSet::new(); + for descriptor in &manifest.layers { + if let Some(title) = descriptor + .annotations + .as_ref() + .and_then(|annotations| annotations.get(ORG_OPENCONTAINERS_IMAGE_TITLE)) + { + titles.insert(title.clone()); + } } - } - for title in titles { - let dest = destination.join(&title); - if !dest.exists() { - hardlink_or_copy(&materialized.join(&title), &dest)?; + for title in titles { + self.storage + .export_hmi_file(&manifest_digest, &title, &destination.join(&title)) + .await?; } + + Ok(config_bytes) } +} - Ok(config_bytes) +#[cfg(test)] +mod tests { + use super::*; + use std::sync::atomic::{AtomicU64, Ordering}; + + const TEST_CONFIG: &[u8] = br#"{ + "schemaVersion": 1, + "kind": "MachineImage", + "system": { + "os": "linux", + "name": "test", + "version": "1", + "architecture": "amd64" + }, + "machine": { + "cpu": { "minimum": 1, "default": 1 }, + "ram": { "minimum": 268435456, "default": 268435456 }, + "boot": [{ "protocol": "firmware-disk/bios", "disk": "root" }], + "access": [{ + "type": "ssh", + "port": 22, + "user": "root", + "auth": "empty-password" + }], + "network": { "mode": "dhcp" } + }, + "disks": [{ + "id": "root", + "format": "qcow2", + "path": "root.hmi", + "digest": "sha256:372409142c91c51316a7ab2b055596e145de56d99aa0f5eb110068876cef0be4", + "virtSize": 14, + "diskSize": 14 + }] + }"#; + + fn temporary_directory() -> PathBuf { + static NEXT_ID: AtomicU64 = AtomicU64::new(0); + std::env::temp_dir().join(format!( + "hule-oci-test-{}-{}", + std::process::id(), + NEXT_ID.fetch_add(1, Ordering::Relaxed) + )) + } + + #[test] + fn manager_uses_separate_oci_and_hmi_roots() { + tokio::runtime::Builder::new_current_thread() + .build() + .unwrap() + .block_on(async { + let root = temporary_directory(); + let source = root.join("source"); + let oci_root = root.join("oci"); + let hmi_root = root.join("hmi"); + let destination = root.join("destination"); + fs::create_dir_all(&source).await.unwrap(); + fs::write(source.join("config.json"), TEST_CONFIG) + .await + .unwrap(); + fs::write(source.join("root.hmi"), b"test hmi image") + .await + .unwrap(); + + let storage = Storage::open(&oci_root, &hmi_root).await.unwrap(); + let images = ImageManager::new(storage); + let reference = "example.test/hule:tokio"; + let digest = images.load(&source, Some(reference)).await.unwrap(); + + assert!(images.contains(reference).await.unwrap()); + assert!(fs::try_exists(oci_root.join("index.json")).await.unwrap()); + assert!( + fs::try_exists( + hmi_root + .join(digest.strip_prefix("sha256:").unwrap_or(&digest)) + .join("root.hmi") + ) + .await + .unwrap() + ); + + let config = images.materialize(reference, &destination).await.unwrap(); + assert!(!config.is_empty()); + assert_eq!( + fs::read(destination.join("root.hmi")).await.unwrap(), + b"test hmi image" + ); + + assert!(matches!(images.contains("missing").await, Ok(false))); + assert!(matches!( + images.materialize("missing", &destination).await, + Err(Error::ImageNotFound(reference)) if reference == "missing" + )); + + fs::remove_dir_all(root).await.unwrap(); + }); + } } diff --git a/crates/hule-oci/src/storage.rs b/crates/hule-oci/src/storage.rs new file mode 100644 index 0000000..21dafe7 --- /dev/null +++ b/crates/hule-oci/src/storage.rs @@ -0,0 +1,213 @@ +// SPDX-FileCopyrightText: 2026 Nikolay Govorov +// SPDX-License-Identifier: Apache-2.0 + +use crate::{Error, Result}; +use oci_client::manifest::OciImageIndex; +use std::io::SeekFrom; +use std::path::{Component, Path, PathBuf}; +use tokio::fs::{self, File}; +use tokio::io::{AsyncSeekExt, AsyncWriteExt}; +use tokio::task; + +/// Filesystem storage for OCI content and materialized HMI images. +/// +/// The roots are independent so callers can place content-addressed OCI data +/// and large materialized disks on different filesystems. +#[derive(Clone, Debug)] +pub struct Storage { + oci_root: PathBuf, + hmi_root: PathBuf, +} + +impl Storage { + /// Opens the storage and creates its filesystem layout when necessary. + pub async fn open(oci_root: impl Into, hmi_root: impl Into) -> Result { + let storage = Self { + oci_root: oci_root.into(), + hmi_root: hmi_root.into(), + }; + + fs::create_dir_all(&storage.hmi_root).await?; + fs::create_dir_all(storage.blobs_dir()).await?; + + let layout = storage.oci_root.join("oci-layout"); + if !fs::try_exists(&layout).await? { + fs::write(&layout, br#"{"imageLayoutVersion":"1.0.0"}"#).await?; + } + + Ok(storage) + } + + pub fn oci_root(&self) -> &Path { + &self.oci_root + } + + pub fn hmi_root(&self) -> &Path { + &self.hmi_root + } + + fn blobs_dir(&self) -> PathBuf { + self.oci_root.join("blobs").join("sha256") + } + + fn index_path(&self) -> PathBuf { + self.oci_root.join("index.json") + } + + fn hmi_image_dir(&self, manifest_digest: &str) -> Result { + let digest = safe_component(strip_sha256(manifest_digest), "manifest digest")?; + Ok(self.hmi_root.join(digest)) + } + + fn hmi_file_path(&self, manifest_digest: &str, title: &str) -> Result { + let title = safe_component(title, "image file name")?; + Ok(self.hmi_image_dir(manifest_digest)?.join(title)) + } + + fn blob_path(&self, digest: &str) -> Result { + let digest = safe_component(strip_sha256(digest), "blob digest")?; + Ok(self.blobs_dir().join(digest)) + } + + pub(crate) async fn read_blob(&self, digest: &str) -> Result> { + Ok(fs::read(self.blob_path(digest)?).await?) + } + + /// Writes raw bytes as a content-addressed blob, no-op if already present. + pub(crate) async fn write_blob(&self, data: T) -> Result<(String, T)> + where + T: AsRef<[u8]> + Send + Sync + 'static, + { + let (digest, data) = + task::spawn_blocking(move || (sha256_hex(data.as_ref()), data)).await?; + let path = self.blob_path(&digest)?; + if !fs::try_exists(&path).await? { + let tmp = path.with_extension("tmp"); + fs::write(&tmp, data.as_ref()).await?; + fs::rename(&tmp, &path).await?; + } + Ok((digest, data)) + } + + pub(crate) async fn read_index(&self) -> Result { + let path = self.index_path(); + if !fs::try_exists(&path).await? { + return Ok(OciImageIndex { + schema_version: 2, + media_type: None, + manifests: vec![], + artifact_type: None, + annotations: None, + }); + } + let data = fs::read(&path).await?; + Ok(serde_json::from_slice(&data)?) + } + + pub(crate) async fn write_index(&self, index: &OciImageIndex) -> Result<()> { + let data = serde_json::to_vec_pretty(index)?; + fs::write(self.index_path(), data).await?; + Ok(()) + } + + pub(crate) async fn ensure_hmi_image(&self, manifest_digest: &str) -> Result<()> { + fs::create_dir_all(self.hmi_image_dir(manifest_digest)?).await?; + Ok(()) + } + + pub(crate) async fn cache_hmi_file( + &self, + manifest_digest: &str, + title: &str, + source: &Path, + ) -> Result<()> { + let destination = self.hmi_file_path(manifest_digest, title)?; + if !fs::try_exists(&destination).await? { + Self::hardlink_or_copy(source, &destination).await?; + } + Ok(()) + } + + pub(crate) async fn write_hmi_file( + &self, + manifest_digest: &str, + title: &str, + mut chunks: Vec<(u64, Vec)>, + ) -> Result<()> { + chunks.sort_by_key(|(offset, _)| *offset); + let mut file = File::create(self.hmi_file_path(manifest_digest, title)?).await?; + for (offset, data) in chunks { + file.seek(SeekFrom::Start(offset)).await?; + file.write_all(&data).await?; + } + file.flush().await?; + Ok(()) + } + + pub(crate) async fn export_hmi_file( + &self, + manifest_digest: &str, + title: &str, + destination: &Path, + ) -> Result<()> { + let source = self.hmi_file_path(manifest_digest, title)?; + if !fs::try_exists(destination).await? { + Self::hardlink_or_copy(&source, destination).await?; + } + Ok(()) + } + + async fn hardlink_or_copy(source: &Path, destination: &Path) -> Result<()> { + if fs::hard_link(source, destination).await.is_err() { + fs::copy(source, destination).await?; + } + Ok(()) + } +} + +fn safe_component<'a>(value: &'a str, kind: &'static str) -> Result<&'a str> { + let mut components = Path::new(value).components(); + match (components.next(), components.next()) { + (Some(Component::Normal(_)), None) => Ok(value), + _ => Err(Error::InvalidStoragePath { + kind, + value: value.to_string(), + }), + } +} + +fn strip_sha256(digest: &str) -> &str { + digest.strip_prefix("sha256:").unwrap_or(digest) +} + +fn sha256_hex(data: &[u8]) -> String { + use sha2::{Digest, Sha256}; + let mut hasher = Sha256::new(); + hasher.update(data); + format!("sha256:{:x}", hasher.finalize()) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn paths_cannot_escape_storage_roots() { + let storage = Storage { + oci_root: "oci".into(), + hmi_root: "hmi".into(), + }; + assert!(matches!( + storage.blob_path("sha256:../../outside"), + Err(Error::InvalidStoragePath { .. }) + )); + assert!(matches!( + storage.hmi_image_dir("sha256:../outside"), + Err(Error::InvalidStoragePath { .. }) + )); + assert!(matches!( + storage.hmi_file_path("sha256:digest", "../outside"), + Err(Error::InvalidStoragePath { .. }) + )); + } +} diff --git a/crates/hule/src/main.rs b/crates/hule/src/main.rs index 644545b..cf9d5d9 100644 --- a/crates/hule/src/main.rs +++ b/crates/hule/src/main.rs @@ -1,7 +1,8 @@ // SPDX-FileCopyrightText: 2026 Nikolay Govorov // SPDX-License-Identifier: Apache-2.0 -use std::path::Path; +use std::fmt::Display; +use std::path::{Path, PathBuf}; use std::process::exit; use clap::{Parser, Subcommand}; @@ -60,19 +61,29 @@ enum MachineCommands { }, } -fn die(msg: impl AsRef) -> ! { - eprintln!("hule: {}", msg.as_ref()); +fn die(msg: impl Display) -> ! { + eprintln!("hule: {msg}"); exit(1); } -async fn cmd_run(reference: &str, port: u16) -> R<()> { - if !hule_oci::contains(reference)? { +async fn cmd_run(images: &hule_oci::ImageManager, reference: &str, port: u16) -> R<()> { + if !images + .contains(reference) + .await + .map_err(|error| error.to_string())? + { eprintln!("hule: {reference} not found locally, pulling..."); - hule_oci::pull(reference).await?; + images + .pull(reference) + .await + .map_err(|error| error.to_string())?; } let scratch = std::env::temp_dir().join(format!("hule-run-{}", std::process::id())); - let config = hule_oci::materialize(reference, &scratch)?; + let config = images + .materialize(reference, &scratch) + .await + .map_err(|error| error.to_string())?; let image = MachineImage::from_json(&config).map_err(|e| e.to_string())?; let hv = QemuHypervisor; @@ -125,20 +136,30 @@ async fn cmd_run(reference: &str, port: u16) -> R<()> { async fn main() { let cli = Cli::parse(); + let home = std::env::var("HOME").unwrap_or_else(|_| die("HOME is not set")); + let root = PathBuf::from(home).join(".hule"); + + let storage = hule_oci::Storage::open(root.join("oci"), root.join("hmi")) + .await + .unwrap_or_else(|e| die(e)); + + let images = hule_oci::ImageManager::new(storage); + match &cli.command { Commands::Image { command } => match &command { ImageCommands::Pull { name } => { - let digest = hule_oci::pull(name).await.unwrap_or_else(|e| die(e)); + let digest = images.pull(name).await.unwrap_or_else(|e| die(e)); eprintln!("hule: pulled {} ({digest})", name); } ImageCommands::Push { name } => { - hule_oci::push(name).await.unwrap_or_else(|e| die(e)); + images.push(name).await.unwrap_or_else(|e| die(e)); eprintln!("hule: pushed {}", name); } ImageCommands::Load { reference, image } => { - let digest = hule_oci::import(Path::new(image), reference.as_deref()) + let digest = images + .load(Path::new(image), reference.as_deref()) .await .unwrap_or_else(|e| die(e)); @@ -152,7 +173,9 @@ async fn main() { }, Commands::Machine { command } => match &command { MachineCommands::Run { image, port } => { - cmd_run(image, *port).await.unwrap_or_else(|e| die(e)); + cmd_run(&images, image, *port) + .await + .unwrap_or_else(|e| die(e)); } }, } -- Gilti