mirror of
https://github.com/edera-dev/krata.git
synced 2025-11-04 07:39:39 +00:00
chore: move kratart::image to krataoci and cleanup dependencies
This commit is contained in:
Alex Zenla
70
crates/krataoci/src/cache.rs
Normal file
70
crates/krataoci/src/cache.rs
Normal file
@ -0,0 +1,70 @@
|
||||
use super::compiler::ImageInfo;
|
||||
use anyhow::Result;
|
||||
use log::debug;
|
||||
use oci_spec::image::{ImageConfiguration, ImageManifest};
|
||||
use std::path::{Path, PathBuf};
|
||||
use tokio::fs;
|
||||
|
||||
pub struct ImageCache {
|
||||
cache_dir: PathBuf,
|
||||
}
|
||||
|
||||
impl ImageCache {
|
||||
pub fn new(cache_dir: &Path) -> Result<ImageCache> {
|
||||
Ok(ImageCache {
|
||||
cache_dir: cache_dir.to_path_buf(),
|
||||
})
|
||||
}
|
||||
|
||||
pub async fn recall(&self, digest: &str) -> Result<Option<ImageInfo>> {
|
||||
let mut squashfs_path = self.cache_dir.clone();
|
||||
let mut config_path = self.cache_dir.clone();
|
||||
let mut manifest_path = self.cache_dir.clone();
|
||||
squashfs_path.push(format!("{}.squashfs", digest));
|
||||
manifest_path.push(format!("{}.manifest.json", digest));
|
||||
config_path.push(format!("{}.config.json", digest));
|
||||
Ok(
|
||||
if squashfs_path.exists() && manifest_path.exists() && config_path.exists() {
|
||||
let squashfs_metadata = fs::metadata(&squashfs_path).await?;
|
||||
let manifest_metadata = fs::metadata(&manifest_path).await?;
|
||||
let config_metadata = fs::metadata(&config_path).await?;
|
||||
if squashfs_metadata.is_file()
|
||||
&& manifest_metadata.is_file()
|
||||
&& config_metadata.is_file()
|
||||
{
|
||||
let manifest_text = fs::read_to_string(&manifest_path).await?;
|
||||
let manifest: ImageManifest = serde_json::from_str(&manifest_text)?;
|
||||
let config_text = fs::read_to_string(&config_path).await?;
|
||||
let config: ImageConfiguration = serde_json::from_str(&config_text)?;
|
||||
debug!("cache hit digest={}", digest);
|
||||
Some(ImageInfo::new(squashfs_path.clone(), manifest, config)?)
|
||||
} else {
|
||||
None
|
||||
}
|
||||
} else {
|
||||
debug!("cache miss digest={}", digest);
|
||||
None
|
||||
},
|
||||
)
|
||||
}
|
||||
|
||||
pub async fn store(&self, digest: &str, info: &ImageInfo) -> Result<ImageInfo> {
|
||||
debug!("cache store digest={}", digest);
|
||||
let mut squashfs_path = self.cache_dir.clone();
|
||||
let mut manifest_path = self.cache_dir.clone();
|
||||
let mut config_path = self.cache_dir.clone();
|
||||
squashfs_path.push(format!("{}.squashfs", digest));
|
||||
manifest_path.push(format!("{}.manifest.json", digest));
|
||||
config_path.push(format!("{}.config.json", digest));
|
||||
fs::copy(&info.image_squashfs, &squashfs_path).await?;
|
||||
let manifest_text = serde_json::to_string_pretty(&info.manifest)?;
|
||||
fs::write(&manifest_path, manifest_text).await?;
|
||||
let config_text = serde_json::to_string_pretty(&info.config)?;
|
||||
fs::write(&config_path, config_text).await?;
|
||||
ImageInfo::new(
|
||||
squashfs_path.clone(),
|
||||
info.manifest.clone(),
|
||||
info.config.clone(),
|
||||
)
|
||||
}
|
||||
}
|
||||
382
crates/krataoci/src/compiler.rs
Normal file
382
crates/krataoci/src/compiler.rs
Normal file
@ -0,0 +1,382 @@
|
||||
use crate::cache::ImageCache;
|
||||
use crate::fetch::{OciImageDownloader, OciImageLayer};
|
||||
use crate::name::ImageName;
|
||||
use crate::registry::OciRegistryPlatform;
|
||||
use anyhow::{anyhow, Result};
|
||||
use backhand::compression::Compressor;
|
||||
use backhand::{FilesystemCompressor, FilesystemWriter, NodeHeader};
|
||||
use log::{debug, trace, warn};
|
||||
use oci_spec::image::{ImageConfiguration, ImageManifest};
|
||||
use std::fs::File;
|
||||
use std::io::{ErrorKind, Read};
|
||||
use std::os::unix::fs::{FileTypeExt, MetadataExt, PermissionsExt};
|
||||
use std::path::{Path, PathBuf};
|
||||
use std::pin::Pin;
|
||||
use tokio::fs;
|
||||
use tokio::io::AsyncRead;
|
||||
use tokio_stream::StreamExt;
|
||||
use tokio_tar::{Archive, Entry};
|
||||
use uuid::Uuid;
|
||||
use walkdir::WalkDir;
|
||||
|
||||
pub const IMAGE_SQUASHFS_VERSION: u64 = 2;
|
||||
|
||||
pub struct ImageInfo {
|
||||
pub image_squashfs: PathBuf,
|
||||
pub manifest: ImageManifest,
|
||||
pub config: ImageConfiguration,
|
||||
}
|
||||
|
||||
impl ImageInfo {
|
||||
pub fn new(
|
||||
squashfs: PathBuf,
|
||||
manifest: ImageManifest,
|
||||
config: ImageConfiguration,
|
||||
) -> Result<ImageInfo> {
|
||||
Ok(ImageInfo {
|
||||
image_squashfs: squashfs,
|
||||
manifest,
|
||||
config,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
pub struct ImageCompiler<'a> {
|
||||
cache: &'a ImageCache,
|
||||
seed: Option<PathBuf>,
|
||||
}
|
||||
|
||||
impl ImageCompiler<'_> {
|
||||
pub fn new(cache: &ImageCache, seed: Option<PathBuf>) -> Result<ImageCompiler> {
|
||||
Ok(ImageCompiler { cache, seed })
|
||||
}
|
||||
|
||||
pub async fn compile(&self, image: &ImageName) -> Result<ImageInfo> {
|
||||
debug!("compile image={image}");
|
||||
let mut tmp_dir = std::env::temp_dir().clone();
|
||||
tmp_dir.push(format!("krata-compile-{}", Uuid::new_v4()));
|
||||
|
||||
let mut image_dir = tmp_dir.clone();
|
||||
image_dir.push("image");
|
||||
fs::create_dir_all(&image_dir).await?;
|
||||
|
||||
let mut layer_dir = tmp_dir.clone();
|
||||
layer_dir.push("layer");
|
||||
fs::create_dir_all(&layer_dir).await?;
|
||||
|
||||
let mut squash_file = tmp_dir.clone();
|
||||
squash_file.push("image.squashfs");
|
||||
let info = self
|
||||
.download_and_compile(image, &layer_dir, &image_dir, &squash_file)
|
||||
.await?;
|
||||
fs::remove_dir_all(&tmp_dir).await?;
|
||||
Ok(info)
|
||||
}
|
||||
|
||||
async fn download_and_compile(
|
||||
&self,
|
||||
image: &ImageName,
|
||||
layer_dir: &Path,
|
||||
image_dir: &Path,
|
||||
squash_file: &Path,
|
||||
) -> Result<ImageInfo> {
|
||||
let downloader = OciImageDownloader::new(
|
||||
self.seed.clone(),
|
||||
layer_dir.to_path_buf(),
|
||||
OciRegistryPlatform::current(),
|
||||
);
|
||||
let resolved = downloader.resolve(image.clone()).await?;
|
||||
let cache_key = format!(
|
||||
"manifest={}:squashfs-version={}\n",
|
||||
resolved.digest, IMAGE_SQUASHFS_VERSION
|
||||
);
|
||||
let cache_digest = sha256::digest(cache_key);
|
||||
|
||||
if let Some(cached) = self.cache.recall(&cache_digest).await? {
|
||||
return Ok(cached);
|
||||
}
|
||||
|
||||
let local = downloader.download(resolved).await?;
|
||||
for layer in &local.layers {
|
||||
debug!(
|
||||
"process layer digest={} compression={:?}",
|
||||
&layer.digest, layer.compression,
|
||||
);
|
||||
self.process_layer_whiteout(layer, image_dir).await?;
|
||||
let mut archive = layer.archive().await?;
|
||||
let mut entries = archive.entries()?;
|
||||
while let Some(entry) = entries.next().await {
|
||||
let mut entry = entry?;
|
||||
let path = entry.path()?;
|
||||
let Some(name) = path.file_name() else {
|
||||
return Err(anyhow!("unable to get file name"));
|
||||
};
|
||||
let Some(name) = name.to_str() else {
|
||||
return Err(anyhow!("unable to get file name as string"));
|
||||
};
|
||||
|
||||
if name.starts_with(".wh.") {
|
||||
continue;
|
||||
} else {
|
||||
self.process_write_entry(&mut entry, layer, image_dir)
|
||||
.await?;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
for layer in &local.layers {
|
||||
if layer.path.exists() {
|
||||
fs::remove_file(&layer.path).await?;
|
||||
}
|
||||
}
|
||||
|
||||
self.squash(image_dir, squash_file)?;
|
||||
let info = ImageInfo::new(
|
||||
squash_file.to_path_buf(),
|
||||
local.image.manifest,
|
||||
local.config,
|
||||
)?;
|
||||
self.cache.store(&cache_digest, &info).await
|
||||
}
|
||||
|
||||
async fn process_layer_whiteout(&self, layer: &OciImageLayer, image_dir: &Path) -> Result<()> {
|
||||
let mut archive = layer.archive().await?;
|
||||
let mut entries = archive.entries()?;
|
||||
while let Some(entry) = entries.next().await {
|
||||
let entry = entry?;
|
||||
let path = entry.path()?;
|
||||
let Some(name) = path.file_name() else {
|
||||
return Err(anyhow!("unable to get file name"));
|
||||
};
|
||||
let Some(name) = name.to_str() else {
|
||||
return Err(anyhow!("unable to get file name as string"));
|
||||
};
|
||||
|
||||
if name.starts_with(".wh.") {
|
||||
self.process_whiteout_entry(&entry, name, layer, image_dir)
|
||||
.await?;
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn process_whiteout_entry(
|
||||
&self,
|
||||
entry: &Entry<Archive<Pin<Box<dyn AsyncRead + Send>>>>,
|
||||
name: &str,
|
||||
layer: &OciImageLayer,
|
||||
image_dir: &Path,
|
||||
) -> Result<()> {
|
||||
let dst = self.check_safe_entry(entry, image_dir)?;
|
||||
let mut dst = dst.clone();
|
||||
dst.pop();
|
||||
|
||||
let opaque = name == ".wh..wh..opq";
|
||||
|
||||
if !opaque {
|
||||
dst.push(&name[4..]);
|
||||
self.check_safe_path(&dst, image_dir)?;
|
||||
}
|
||||
|
||||
trace!(
|
||||
"whiteout entry layer={} path={:?}",
|
||||
&layer.digest,
|
||||
entry.path()?
|
||||
);
|
||||
|
||||
if opaque {
|
||||
if dst.is_dir() {
|
||||
let mut reader = fs::read_dir(dst).await?;
|
||||
while let Some(entry) = reader.next_entry().await? {
|
||||
let path = entry.path();
|
||||
if path.is_symlink() || path.is_file() {
|
||||
fs::remove_file(&path).await?;
|
||||
} else if path.is_dir() {
|
||||
fs::remove_dir_all(&path).await?;
|
||||
} else {
|
||||
return Err(anyhow!("opaque whiteout entry did not exist"));
|
||||
}
|
||||
}
|
||||
} else {
|
||||
warn!(
|
||||
"whiteout opaque entry missing locally layer={} path={:?} local={:?}",
|
||||
&layer.digest,
|
||||
entry.path()?,
|
||||
dst,
|
||||
);
|
||||
}
|
||||
} else if dst.is_file() || dst.is_symlink() {
|
||||
fs::remove_file(&dst).await?;
|
||||
} else if dst.is_dir() {
|
||||
fs::remove_dir_all(&dst).await?;
|
||||
} else {
|
||||
warn!(
|
||||
"whiteout entry missing locally layer={} path={:?} local={:?}",
|
||||
&layer.digest,
|
||||
entry.path()?,
|
||||
dst,
|
||||
);
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn process_write_entry(
|
||||
&self,
|
||||
entry: &mut Entry<Archive<Pin<Box<dyn AsyncRead + Send>>>>,
|
||||
layer: &OciImageLayer,
|
||||
image_dir: &Path,
|
||||
) -> Result<()> {
|
||||
let uid = entry.header().uid()?;
|
||||
let gid = entry.header().gid()?;
|
||||
trace!(
|
||||
"unpack entry layer={} path={:?} type={:?} uid={} gid={}",
|
||||
&layer.digest,
|
||||
entry.path()?,
|
||||
entry.header().entry_type(),
|
||||
uid,
|
||||
gid,
|
||||
);
|
||||
entry.set_preserve_mtime(true);
|
||||
entry.set_preserve_permissions(true);
|
||||
entry.set_unpack_xattrs(true);
|
||||
if let Some(path) = entry.unpack_in(image_dir).await? {
|
||||
if !path.is_symlink() {
|
||||
std::os::unix::fs::chown(path, Some(uid as u32), Some(gid as u32))?;
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn check_safe_entry(
|
||||
&self,
|
||||
entry: &Entry<Archive<Pin<Box<dyn AsyncRead + Send>>>>,
|
||||
image_dir: &Path,
|
||||
) -> Result<PathBuf> {
|
||||
let mut dst = image_dir.to_path_buf();
|
||||
dst.push(entry.path()?);
|
||||
if let Some(name) = dst.file_name() {
|
||||
if let Some(name) = name.to_str() {
|
||||
if name.starts_with(".wh.") {
|
||||
let copy = dst.clone();
|
||||
dst.pop();
|
||||
self.check_safe_path(&dst, image_dir)?;
|
||||
return Ok(copy);
|
||||
}
|
||||
}
|
||||
}
|
||||
self.check_safe_path(&dst, image_dir)?;
|
||||
Ok(dst)
|
||||
}
|
||||
|
||||
fn check_safe_path(&self, dst: &Path, image_dir: &Path) -> Result<()> {
|
||||
let resolved = path_clean::clean(dst);
|
||||
if !resolved.starts_with(image_dir) {
|
||||
return Err(anyhow!("layer attempts to work outside image dir"));
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn squash(&self, image_dir: &Path, squash_file: &Path) -> Result<()> {
|
||||
let mut writer = FilesystemWriter::default();
|
||||
writer.set_compressor(FilesystemCompressor::new(Compressor::Gzip, None)?);
|
||||
let walk = WalkDir::new(image_dir).follow_links(false);
|
||||
for entry in walk {
|
||||
let entry = entry?;
|
||||
let rel = entry
|
||||
.path()
|
||||
.strip_prefix(image_dir)?
|
||||
.to_str()
|
||||
.ok_or_else(|| anyhow!("failed to strip prefix of tmpdir"))?;
|
||||
let rel = format!("/{}", rel);
|
||||
trace!("squash write {}", rel);
|
||||
let typ = entry.file_type();
|
||||
let metadata = std::fs::symlink_metadata(entry.path())?;
|
||||
let uid = metadata.uid();
|
||||
let gid = metadata.gid();
|
||||
let mode = metadata.permissions().mode();
|
||||
let mtime = metadata.mtime();
|
||||
|
||||
if rel == "/" {
|
||||
writer.set_root_uid(uid);
|
||||
writer.set_root_gid(gid);
|
||||
writer.set_root_mode(mode as u16);
|
||||
continue;
|
||||
}
|
||||
|
||||
let header = NodeHeader {
|
||||
permissions: mode as u16,
|
||||
uid,
|
||||
gid,
|
||||
mtime: mtime as u32,
|
||||
};
|
||||
if typ.is_symlink() {
|
||||
let symlink = std::fs::read_link(entry.path())?;
|
||||
let symlink = symlink
|
||||
.to_str()
|
||||
.ok_or_else(|| anyhow!("failed to read symlink"))?;
|
||||
writer.push_symlink(symlink, rel, header)?;
|
||||
} else if typ.is_dir() {
|
||||
writer.push_dir(rel, header)?;
|
||||
} else if typ.is_file() {
|
||||
writer.push_file(ConsumingFileReader::new(entry.path()), rel, header)?;
|
||||
} else if typ.is_block_device() {
|
||||
let device = metadata.dev();
|
||||
writer.push_block_device(device as u32, rel, header)?;
|
||||
} else if typ.is_char_device() {
|
||||
let device = metadata.dev();
|
||||
writer.push_char_device(device as u32, rel, header)?;
|
||||
} else {
|
||||
return Err(anyhow!("invalid file type"));
|
||||
}
|
||||
}
|
||||
|
||||
let squash_file_path = squash_file
|
||||
.to_str()
|
||||
.ok_or_else(|| anyhow!("failed to convert squashfs string"))?;
|
||||
|
||||
let mut file = File::create(squash_file)?;
|
||||
trace!("squash generate: {}", squash_file_path);
|
||||
writer.write(&mut file)?;
|
||||
std::fs::remove_dir_all(image_dir)?;
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
struct ConsumingFileReader {
|
||||
path: PathBuf,
|
||||
file: Option<File>,
|
||||
}
|
||||
|
||||
impl ConsumingFileReader {
|
||||
fn new(path: &Path) -> ConsumingFileReader {
|
||||
ConsumingFileReader {
|
||||
path: path.to_path_buf(),
|
||||
file: None,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl Read for ConsumingFileReader {
|
||||
fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
|
||||
if self.file.is_none() {
|
||||
self.file = Some(File::open(&self.path)?);
|
||||
}
|
||||
let Some(ref mut file) = self.file else {
|
||||
return Err(std::io::Error::new(
|
||||
ErrorKind::NotFound,
|
||||
"file was not opened",
|
||||
));
|
||||
};
|
||||
file.read(buf)
|
||||
}
|
||||
}
|
||||
|
||||
impl Drop for ConsumingFileReader {
|
||||
fn drop(&mut self) {
|
||||
let file = self.file.take();
|
||||
drop(file);
|
||||
if let Err(error) = std::fs::remove_file(&self.path) {
|
||||
warn!("failed to delete consuming file {:?}: {}", self.path, error);
|
||||
}
|
||||
}
|
||||
}
|
||||
280
crates/krataoci/src/fetch.rs
Normal file
280
crates/krataoci/src/fetch.rs
Normal file
@ -0,0 +1,280 @@
|
||||
use super::{
|
||||
name::ImageName,
|
||||
registry::{OciRegistryClient, OciRegistryPlatform},
|
||||
};
|
||||
|
||||
use std::{
|
||||
path::{Path, PathBuf},
|
||||
pin::Pin,
|
||||
};
|
||||
|
||||
use anyhow::{anyhow, Result};
|
||||
use async_compression::tokio::bufread::{GzipDecoder, ZstdDecoder};
|
||||
use log::debug;
|
||||
use oci_spec::image::{
|
||||
Descriptor, ImageConfiguration, ImageIndex, ImageManifest, MediaType, ToDockerV2S2,
|
||||
};
|
||||
use serde::de::DeserializeOwned;
|
||||
use tokio::{
|
||||
fs::File,
|
||||
io::{AsyncRead, AsyncReadExt, BufReader},
|
||||
};
|
||||
use tokio_stream::StreamExt;
|
||||
use tokio_tar::Archive;
|
||||
|
||||
pub struct OciImageDownloader {
|
||||
seed: Option<PathBuf>,
|
||||
storage: PathBuf,
|
||||
platform: OciRegistryPlatform,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, PartialEq, Eq)]
|
||||
pub enum OciImageLayerCompression {
|
||||
None,
|
||||
Gzip,
|
||||
Zstd,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug)]
|
||||
pub struct OciImageLayer {
|
||||
pub path: PathBuf,
|
||||
pub digest: String,
|
||||
pub compression: OciImageLayerCompression,
|
||||
}
|
||||
|
||||
impl OciImageLayer {
|
||||
pub async fn decompress(&self) -> Result<Pin<Box<dyn AsyncRead + Send>>> {
|
||||
let file = File::open(&self.path).await?;
|
||||
let reader = BufReader::new(file);
|
||||
let reader: Pin<Box<dyn AsyncRead + Send>> = match self.compression {
|
||||
OciImageLayerCompression::None => Box::pin(reader),
|
||||
OciImageLayerCompression::Gzip => Box::pin(GzipDecoder::new(reader)),
|
||||
OciImageLayerCompression::Zstd => Box::pin(ZstdDecoder::new(reader)),
|
||||
};
|
||||
Ok(reader)
|
||||
}
|
||||
|
||||
pub async fn archive(&self) -> Result<Archive<Pin<Box<dyn AsyncRead + Send>>>> {
|
||||
let decompress = self.decompress().await?;
|
||||
Ok(Archive::new(decompress))
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug)]
|
||||
pub struct OciResolvedImage {
|
||||
pub name: ImageName,
|
||||
pub digest: String,
|
||||
pub manifest: ImageManifest,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug)]
|
||||
pub struct OciLocalImage {
|
||||
pub image: OciResolvedImage,
|
||||
pub config: ImageConfiguration,
|
||||
pub layers: Vec<OciImageLayer>,
|
||||
}
|
||||
|
||||
impl OciImageDownloader {
|
||||
pub fn new(
|
||||
seed: Option<PathBuf>,
|
||||
storage: PathBuf,
|
||||
platform: OciRegistryPlatform,
|
||||
) -> OciImageDownloader {
|
||||
OciImageDownloader {
|
||||
seed,
|
||||
storage,
|
||||
platform,
|
||||
}
|
||||
}
|
||||
|
||||
async fn load_seed_json_blob<T: DeserializeOwned>(
|
||||
&self,
|
||||
descriptor: &Descriptor,
|
||||
) -> Result<Option<T>> {
|
||||
let digest = descriptor.digest();
|
||||
let Some((digest_type, digest_content)) = digest.split_once(':') else {
|
||||
return Err(anyhow!("digest content was not properly formatted"));
|
||||
};
|
||||
let want = format!("blobs/{}/{}", digest_type, digest_content);
|
||||
self.load_seed_json(&want).await
|
||||
}
|
||||
|
||||
async fn load_seed_json<T: DeserializeOwned>(&self, want: &str) -> Result<Option<T>> {
|
||||
let Some(ref seed) = self.seed else {
|
||||
return Ok(None);
|
||||
};
|
||||
|
||||
let file = File::open(seed).await?;
|
||||
let mut archive = Archive::new(file);
|
||||
let mut entries = archive.entries()?;
|
||||
while let Some(entry) = entries.next().await {
|
||||
let mut entry = entry?;
|
||||
let path = String::from_utf8(entry.path_bytes().to_vec())?;
|
||||
if path == want {
|
||||
let mut content = String::new();
|
||||
entry.read_to_string(&mut content).await?;
|
||||
let data = serde_json::from_str::<T>(&content)?;
|
||||
return Ok(Some(data));
|
||||
}
|
||||
}
|
||||
Ok(None)
|
||||
}
|
||||
|
||||
async fn extract_seed_blob(&self, descriptor: &Descriptor, to: &Path) -> Result<bool> {
|
||||
let Some(ref seed) = self.seed else {
|
||||
return Ok(false);
|
||||
};
|
||||
|
||||
let digest = descriptor.digest();
|
||||
let Some((digest_type, digest_content)) = digest.split_once(':') else {
|
||||
return Err(anyhow!("digest content was not properly formatted"));
|
||||
};
|
||||
let want = format!("blobs/{}/{}", digest_type, digest_content);
|
||||
|
||||
let seed = File::open(seed).await?;
|
||||
let mut archive = Archive::new(seed);
|
||||
let mut entries = archive.entries()?;
|
||||
while let Some(entry) = entries.next().await {
|
||||
let mut entry = entry?;
|
||||
let path = String::from_utf8(entry.path_bytes().to_vec())?;
|
||||
if path == want {
|
||||
let mut file = File::create(to).await?;
|
||||
tokio::io::copy(&mut entry, &mut file).await?;
|
||||
return Ok(true);
|
||||
}
|
||||
}
|
||||
Ok(false)
|
||||
}
|
||||
|
||||
pub async fn resolve(&self, image: ImageName) -> Result<OciResolvedImage> {
|
||||
debug!("resolve manifest image={}", image);
|
||||
|
||||
if let Some(index) = self.load_seed_json::<ImageIndex>("index.json").await? {
|
||||
let mut found: Option<&Descriptor> = None;
|
||||
for manifest in index.manifests() {
|
||||
let Some(annotations) = manifest.annotations() else {
|
||||
continue;
|
||||
};
|
||||
|
||||
let mut image_name = annotations.get("io.containerd.image.name");
|
||||
if image_name.is_none() {
|
||||
image_name = annotations.get("org.opencontainers.image.ref.name");
|
||||
}
|
||||
|
||||
let Some(image_name) = image_name else {
|
||||
continue;
|
||||
};
|
||||
|
||||
if *image_name != image.to_string() {
|
||||
continue;
|
||||
}
|
||||
|
||||
if let Some(platform) = manifest.platform() {
|
||||
if *platform.architecture() != self.platform.arch
|
||||
|| *platform.os() != self.platform.os
|
||||
{
|
||||
continue;
|
||||
}
|
||||
}
|
||||
found = Some(manifest);
|
||||
break;
|
||||
}
|
||||
|
||||
if let Some(found) = found {
|
||||
if let Some(manifest) = self.load_seed_json_blob(found).await? {
|
||||
debug!(
|
||||
"found seeded manifest image={} manifest={}",
|
||||
image,
|
||||
found.digest()
|
||||
);
|
||||
return Ok(OciResolvedImage {
|
||||
name: image,
|
||||
digest: found.digest().clone(),
|
||||
manifest,
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
let mut client = OciRegistryClient::new(image.registry_url()?, self.platform.clone())?;
|
||||
let (manifest, digest) = client
|
||||
.get_manifest_with_digest(&image.name, &image.reference)
|
||||
.await?;
|
||||
Ok(OciResolvedImage {
|
||||
name: image,
|
||||
digest,
|
||||
manifest,
|
||||
})
|
||||
}
|
||||
|
||||
pub async fn download(&self, image: OciResolvedImage) -> Result<OciLocalImage> {
|
||||
let config: ImageConfiguration;
|
||||
|
||||
let mut client = OciRegistryClient::new(image.name.registry_url()?, self.platform.clone())?;
|
||||
if let Some(seeded) = self
|
||||
.load_seed_json_blob::<ImageConfiguration>(image.manifest.config())
|
||||
.await?
|
||||
{
|
||||
config = seeded;
|
||||
} else {
|
||||
let config_bytes = client
|
||||
.get_blob(&image.name.name, image.manifest.config())
|
||||
.await?;
|
||||
config = serde_json::from_slice(&config_bytes)?;
|
||||
}
|
||||
let mut layers = Vec::new();
|
||||
for layer in image.manifest.layers() {
|
||||
layers.push(self.acquire_layer(&image.name, layer, &mut client).await?);
|
||||
}
|
||||
Ok(OciLocalImage {
|
||||
image,
|
||||
config,
|
||||
layers,
|
||||
})
|
||||
}
|
||||
|
||||
async fn acquire_layer(
|
||||
&self,
|
||||
image: &ImageName,
|
||||
layer: &Descriptor,
|
||||
client: &mut OciRegistryClient,
|
||||
) -> Result<OciImageLayer> {
|
||||
debug!(
|
||||
"acquire layer digest={} size={}",
|
||||
layer.digest(),
|
||||
layer.size()
|
||||
);
|
||||
let mut layer_path = self.storage.clone();
|
||||
layer_path.push(format!("{}.layer", layer.digest()));
|
||||
|
||||
let seeded = self.extract_seed_blob(layer, &layer_path).await?;
|
||||
if !seeded {
|
||||
let file = File::create(&layer_path).await?;
|
||||
let size = client.write_blob_to_file(&image.name, layer, file).await?;
|
||||
if layer.size() as u64 != size {
|
||||
return Err(anyhow!(
|
||||
"downloaded layer size differs from size in manifest",
|
||||
));
|
||||
}
|
||||
}
|
||||
|
||||
let mut media_type = layer.media_type().clone();
|
||||
|
||||
// docker layer compatibility
|
||||
if media_type.to_string() == MediaType::ImageLayerGzip.to_docker_v2s2()? {
|
||||
media_type = MediaType::ImageLayerGzip;
|
||||
}
|
||||
|
||||
let compression = match media_type {
|
||||
MediaType::ImageLayer => OciImageLayerCompression::None,
|
||||
MediaType::ImageLayerGzip => OciImageLayerCompression::Gzip,
|
||||
MediaType::ImageLayerZstd => OciImageLayerCompression::Zstd,
|
||||
other => return Err(anyhow!("found layer with unknown media type: {}", other)),
|
||||
};
|
||||
Ok(OciImageLayer {
|
||||
path: layer_path,
|
||||
digest: layer.digest().clone(),
|
||||
compression,
|
||||
})
|
||||
}
|
||||
}
|
||||
5
crates/krataoci/src/lib.rs
Normal file
5
crates/krataoci/src/lib.rs
Normal file
@ -0,0 +1,5 @@
|
||||
pub mod cache;
|
||||
pub mod compiler;
|
||||
pub mod fetch;
|
||||
pub mod name;
|
||||
pub mod registry;
|
||||
88
crates/krataoci/src/name.rs
Normal file
88
crates/krataoci/src/name.rs
Normal file
@ -0,0 +1,88 @@
|
||||
use anyhow::Result;
|
||||
use std::fmt;
|
||||
use url::Url;
|
||||
|
||||
const DOCKER_HUB_MIRROR: &str = "mirror.gcr.io";
|
||||
const DEFAULT_IMAGE_TAG: &str = "latest";
|
||||
|
||||
#[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 full_name = name.to_string();
|
||||
let name = full_name.clone();
|
||||
let (mut hostname, mut name) = name
|
||||
.split_once('/')
|
||||
.map(|x| (x.0.to_string(), x.1.to_string()))
|
||||
.unwrap_or_else(|| (DOCKER_HUB_MIRROR.to_string(), format!("library/{}", name)));
|
||||
|
||||
// heuristic to find any docker hub image formats
|
||||
// that may be in the hostname format. for example:
|
||||
// abc/xyz:latest will trigger this if check, but abc.io/xyz:latest will not,
|
||||
// and neither will abc/hello/xyz:latest
|
||||
if !hostname.contains('.') && full_name.chars().filter(|x| *x == '/').count() == 1 {
|
||||
name = format!("{}/{}", hostname, name);
|
||||
hostname = DOCKER_HUB_MIRROR.to_string();
|
||||
}
|
||||
|
||||
let (hostname, port) = if let Some((hostname, port)) = hostname
|
||||
.split_once(':')
|
||||
.map(|x| (x.0.to_string(), x.1.to_string()))
|
||||
{
|
||||
(hostname, Some(str::parse(&port)?))
|
||||
} else {
|
||||
(hostname, None)
|
||||
};
|
||||
let (name, reference) = name
|
||||
.split_once(':')
|
||||
.map(|x| (x.0.to_string(), x.1.to_string()))
|
||||
.unwrap_or((name.to_string(), DEFAULT_IMAGE_TAG.to_string()));
|
||||
Ok(ImageName {
|
||||
hostname,
|
||||
port,
|
||||
name,
|
||||
reference,
|
||||
})
|
||||
}
|
||||
|
||||
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)?)
|
||||
}
|
||||
}
|
||||
244
crates/krataoci/src/registry.rs
Normal file
244
crates/krataoci/src/registry.rs
Normal file
@ -0,0 +1,244 @@
|
||||
use std::collections::HashMap;
|
||||
|
||||
use anyhow::{anyhow, Result};
|
||||
use bytes::Bytes;
|
||||
use oci_spec::image::{Arch, Descriptor, ImageIndex, ImageManifest, MediaType, Os, ToDockerV2S2};
|
||||
use reqwest::{Client, RequestBuilder, Response, StatusCode};
|
||||
use tokio::{fs::File, io::AsyncWriteExt};
|
||||
use url::Url;
|
||||
|
||||
#[derive(Clone, Debug)]
|
||||
pub struct OciRegistryPlatform {
|
||||
pub os: Os,
|
||||
pub arch: Arch,
|
||||
}
|
||||
|
||||
impl OciRegistryPlatform {
|
||||
#[cfg(target_arch = "x86_64")]
|
||||
const CURRENT_ARCH: Arch = Arch::Amd64;
|
||||
#[cfg(target_arch = "aarch64")]
|
||||
const CURRENT_ARCH: Arch = Arch::ARM64;
|
||||
|
||||
pub fn new(os: Os, arch: Arch) -> OciRegistryPlatform {
|
||||
OciRegistryPlatform { os, arch }
|
||||
}
|
||||
|
||||
pub fn current() -> OciRegistryPlatform {
|
||||
OciRegistryPlatform {
|
||||
os: Os::Linux,
|
||||
arch: OciRegistryPlatform::CURRENT_ARCH,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
pub struct OciRegistryClient {
|
||||
agent: Client,
|
||||
url: Url,
|
||||
platform: OciRegistryPlatform,
|
||||
token: Option<String>,
|
||||
}
|
||||
|
||||
impl OciRegistryClient {
|
||||
pub fn new(url: Url, platform: OciRegistryPlatform) -> Result<OciRegistryClient> {
|
||||
Ok(OciRegistryClient {
|
||||
agent: Client::new(),
|
||||
url,
|
||||
platform,
|
||||
token: None,
|
||||
})
|
||||
}
|
||||
|
||||
async fn call(&mut self, mut req: RequestBuilder) -> Result<Response> {
|
||||
if let Some(ref token) = self.token {
|
||||
req = req.bearer_auth(token);
|
||||
}
|
||||
let req_first_try = req.try_clone().ok_or(anyhow!("request is not clonable"))?;
|
||||
let response = self.agent.execute(req_first_try.build()?).await?;
|
||||
if response.status() == StatusCode::UNAUTHORIZED && self.token.is_none() {
|
||||
let Some(www_authenticate) = response.headers().get("www-authenticate") else {
|
||||
return Err(anyhow!("not authorized to perform this action"));
|
||||
};
|
||||
|
||||
let www_authenticate = www_authenticate.to_str()?;
|
||||
if !www_authenticate.starts_with("Bearer ") {
|
||||
return Err(anyhow!("unknown authentication scheme"));
|
||||
}
|
||||
|
||||
let details = &www_authenticate[7..];
|
||||
let details = details
|
||||
.split(',')
|
||||
.map(|x| x.split('='))
|
||||
.map(|mut x| (x.next(), x.next()))
|
||||
.filter(|(key, value)| key.is_some() && value.is_some())
|
||||
.map(|(key, value)| {
|
||||
(
|
||||
key.unwrap().trim().to_lowercase(),
|
||||
value.unwrap().trim().to_string(),
|
||||
)
|
||||
})
|
||||
.map(|(key, value)| (key, value.trim_matches('\"').to_string()))
|
||||
.collect::<HashMap<_, _>>();
|
||||
let realm = details.get("realm");
|
||||
let service = details.get("service");
|
||||
let scope = details.get("scope");
|
||||
if realm.is_none() || service.is_none() || scope.is_none() {
|
||||
return Err(anyhow!(
|
||||
"unknown authentication scheme: realm, service, and scope are required"
|
||||
));
|
||||
}
|
||||
let mut url = Url::parse(realm.unwrap())?;
|
||||
url.query_pairs_mut()
|
||||
.append_pair("service", service.unwrap())
|
||||
.append_pair("scope", scope.unwrap());
|
||||
let token_response = self.agent.get(url.clone()).send().await?;
|
||||
if token_response.status() != StatusCode::OK {
|
||||
return Err(anyhow!(
|
||||
"failed to acquire token via {}: status {}",
|
||||
url,
|
||||
token_response.status()
|
||||
));
|
||||
}
|
||||
let token_bytes = token_response.bytes().await?;
|
||||
let token = serde_json::from_slice::<serde_json::Value>(&token_bytes)?;
|
||||
let token = token
|
||||
.get("token")
|
||||
.and_then(|x| x.as_str())
|
||||
.ok_or(anyhow!("token key missing from response"))?;
|
||||
self.token = Some(token.to_string());
|
||||
return Ok(self.agent.execute(req.bearer_auth(token).build()?).await?);
|
||||
}
|
||||
|
||||
if !response.status().is_success() {
|
||||
return Err(anyhow!(
|
||||
"request to {} failed: status {}",
|
||||
req.build()?.url(),
|
||||
response.status()
|
||||
));
|
||||
}
|
||||
|
||||
Ok(response)
|
||||
}
|
||||
|
||||
pub async fn get_blob<N: AsRef<str>>(
|
||||
&mut self,
|
||||
name: N,
|
||||
descriptor: &Descriptor,
|
||||
) -> Result<Bytes> {
|
||||
let url = self.url.join(&format!(
|
||||
"/v2/{}/blobs/{}",
|
||||
name.as_ref(),
|
||||
descriptor.digest()
|
||||
))?;
|
||||
let response = self.call(self.agent.get(url.as_str())).await?;
|
||||
Ok(response.bytes().await?)
|
||||
}
|
||||
|
||||
pub async fn write_blob_to_file<N: AsRef<str>>(
|
||||
&mut self,
|
||||
name: N,
|
||||
descriptor: &Descriptor,
|
||||
mut dest: File,
|
||||
) -> Result<u64> {
|
||||
let url = self.url.join(&format!(
|
||||
"/v2/{}/blobs/{}",
|
||||
name.as_ref(),
|
||||
descriptor.digest()
|
||||
))?;
|
||||
let mut response = self.call(self.agent.get(url.as_str())).await?;
|
||||
let mut size: u64 = 0;
|
||||
while let Some(chunk) = response.chunk().await? {
|
||||
dest.write_all(&chunk).await?;
|
||||
size += chunk.len() as u64;
|
||||
}
|
||||
Ok(size)
|
||||
}
|
||||
|
||||
async fn get_raw_manifest_with_digest<N: AsRef<str>, R: AsRef<str>>(
|
||||
&mut self,
|
||||
name: N,
|
||||
reference: R,
|
||||
) -> Result<(ImageManifest, String)> {
|
||||
let url = self.url.join(&format!(
|
||||
"/v2/{}/manifests/{}",
|
||||
name.as_ref(),
|
||||
reference.as_ref()
|
||||
))?;
|
||||
let accept = format!(
|
||||
"{}, {}, {}, {}",
|
||||
MediaType::ImageManifest.to_docker_v2s2()?,
|
||||
MediaType::ImageManifest,
|
||||
MediaType::ImageIndex,
|
||||
MediaType::ImageIndex.to_docker_v2s2()?,
|
||||
);
|
||||
let response = self
|
||||
.call(self.agent.get(url.as_str()).header("Accept", &accept))
|
||||
.await?;
|
||||
let digest = response
|
||||
.headers()
|
||||
.get("Docker-Content-Digest")
|
||||
.ok_or_else(|| anyhow!("fetching manifest did not yield a content digest"))?
|
||||
.to_str()?
|
||||
.to_string();
|
||||
let manifest = serde_json::from_str(&response.text().await?)?;
|
||||
Ok((manifest, digest))
|
||||
}
|
||||
|
||||
pub async fn get_manifest_with_digest<N: AsRef<str>, R: AsRef<str>>(
|
||||
&mut self,
|
||||
name: N,
|
||||
reference: R,
|
||||
) -> Result<(ImageManifest, String)> {
|
||||
let url = self.url.join(&format!(
|
||||
"/v2/{}/manifests/{}",
|
||||
name.as_ref(),
|
||||
reference.as_ref()
|
||||
))?;
|
||||
let accept = format!(
|
||||
"{}, {}, {}, {}",
|
||||
MediaType::ImageManifest.to_docker_v2s2()?,
|
||||
MediaType::ImageManifest,
|
||||
MediaType::ImageIndex,
|
||||
MediaType::ImageIndex.to_docker_v2s2()?,
|
||||
);
|
||||
let response = self
|
||||
.call(self.agent.get(url.as_str()).header("Accept", &accept))
|
||||
.await?;
|
||||
let content_type = response
|
||||
.headers()
|
||||
.get("Content-Type")
|
||||
.ok_or_else(|| anyhow!("registry response did not have a Content-Type header"))?
|
||||
.to_str()?;
|
||||
if content_type == MediaType::ImageIndex.to_string()
|
||||
|| content_type == MediaType::ImageIndex.to_docker_v2s2()?
|
||||
{
|
||||
let index = serde_json::from_str(&response.text().await?)?;
|
||||
let descriptor = self
|
||||
.pick_manifest(index)
|
||||
.ok_or_else(|| anyhow!("unable to pick manifest from index"))?;
|
||||
return self
|
||||
.get_raw_manifest_with_digest(name, descriptor.digest())
|
||||
.await;
|
||||
}
|
||||
let digest = response
|
||||
.headers()
|
||||
.get("Docker-Content-Digest")
|
||||
.ok_or_else(|| anyhow!("fetching manifest did not yield a content digest"))?
|
||||
.to_str()?
|
||||
.to_string();
|
||||
let manifest = serde_json::from_str(&response.text().await?)?;
|
||||
Ok((manifest, digest))
|
||||
}
|
||||
|
||||
fn pick_manifest(&mut self, index: ImageIndex) -> Option<Descriptor> {
|
||||
for item in index.manifests() {
|
||||
if let Some(platform) = item.platform() {
|
||||
if *platform.os() == self.platform.os
|
||||
&& *platform.architecture() == self.platform.arch
|
||||
{
|
||||
return Some(item.clone());
|
||||
}
|
||||
}
|
||||
}
|
||||
None
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user