use {
crate::{
comp::{comp_reader, is_comp_ext, is_tar_ext, tar_reader},
content::ContentProvider,
hash::{AsyncHashingRead, Hash, HashAlgo, Hashable, HashingReader},
is_url,
staging::{FileList, Stage, StagingFile, StagingFileSystem},
tar,
},
clap::Args,
rustix::{
fd::{AsRawFd, OwnedFd},
fs::{fstat, openat, FileType, Mode, OFlags, CWD},
},
serde::{Deserialize, Serialize},
smol::{io::AsyncRead, stream::StreamExt},
std::{
borrow::Cow,
ffi::OsStr,
future::Future,
io,
num::NonZero,
path::{Path, PathBuf},
pin::{pin, Pin},
},
};
#[derive(Clone)]
pub(crate) struct FileModeArgParser;
impl clap::builder::TypedValueParser for FileModeArgParser {
type Value = NonZero<u32>;
fn parse_ref(
&self,
cmd: &clap::Command,
arg: Option<&clap::Arg>,
value: &OsStr,
) -> Result<Self::Value, clap::Error> {
let value = value.to_str().ok_or_else(|| {
let mut err = clap::Error::new(clap::error::ErrorKind::InvalidUtf8).with_cmd(cmd);
if let Some(arg) = arg {
err.insert(
clap::error::ContextKind::InvalidArg,
clap::error::ContextValue::String(arg.to_string()),
);
}
err
})?;
parse_file_mode(value).map_err(|message| {
let mut err = clap::Error::new(clap::error::ErrorKind::ValueValidation).with_cmd(cmd);
if let Some(arg) = arg {
err.insert(
clap::error::ContextKind::InvalidArg,
clap::error::ContextValue::String(arg.to_string()),
);
}
err.insert(
clap::error::ContextKind::InvalidValue,
clap::error::ContextValue::String(value.to_string()),
);
err.insert(
clap::error::ContextKind::Custom,
clap::error::ContextValue::String(message),
);
err
})
}
}
pub(crate) fn parse_file_mode(value: &str) -> Result<NonZero<u32>, String> {
let value = value.trim();
if value.is_empty() {
return Err("mode cannot be empty".to_string());
}
let digits = if let Some(rest) = value.strip_prefix("0o").or(value.strip_prefix("0O")) {
rest
} else if value.chars().all(|c| c.is_ascii_digit()) {
if value.chars().any(|c| c == '8' || c == '9') {
return Err("mode must be octal (digits 0-7), e.g. 644 or 755".to_string());
}
value
} else {
return Err("mode must be an octal number like 644 or 755".to_string());
};
let digits = digits.trim();
if digits.is_empty() {
return Err("mode digits are missing".to_string());
}
let parsed =
u32::from_str_radix(digits, 8).map_err(|err| format!("invalid mode '{value}': {err}"))?;
if parsed == 0 {
return Err("mode cannot be zero".to_string());
}
Ok(NonZero::new(parsed).expect("mode is non-zero"))
}
fn normalize_target_absolute(target: &str) -> io::Result<Cow<'_, str>> {
if target.is_empty() {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
"target path cannot be empty",
));
}
if target.starts_with('/') {
Ok(target.into())
} else {
Ok(format!("/{target}").into())
}
}
fn stage_file_target<'a>(target: &'a str, uri: &str) -> io::Result<Cow<'a, str>> {
let target = normalize_target_absolute(target)?;
if target.ends_with('/') {
let name = artifact_file_name(uri).ok_or_else(|| {
io::Error::new(
io::ErrorKind::InvalidInput,
format!("cannot derive file name from artifact URI: {uri}"),
)
})?;
Ok(format!("{target}{name}").into())
} else {
Ok(target)
}
}
fn stage_text_target(target: &str, name: &str) -> io::Result<String> {
let target = normalize_target_absolute(target)?;
if target.ends_with('/') {
Ok(format!("{target}{name}"))
} else {
Ok(target.into_owned())
}
}
fn stage_dir_target(target: &str) -> io::Result<String> {
normalize_target_absolute(target).map(Cow::into_owned)
}
fn artifact_file_name(uri: &str) -> Option<&str> {
let uri = uri.split(['?', '#']).next().unwrap_or(uri);
let uri = crate::strip_url_scheme(uri);
uri.rsplit('/').find(|s| !s.is_empty())
}
fn validate_unpacked_path(path: &str) -> io::Result<()> {
let path = Path::new(path);
if path.is_absolute() {
return Err(io::Error::new(
io::ErrorKind::InvalidData,
format!("archive entry path must be relative: {}", path.display()),
));
}
if path
.components()
.any(|c| matches!(c, std::path::Component::ParentDir))
{
return Err(io::Error::new(
io::ErrorKind::InvalidData,
format!(
"archive entry path must not contain '..': {}",
path.display()
),
));
}
Ok(())
}
#[derive(Args)]
pub struct ArtifactArg {
#[arg(long = "mode", value_name = "MODE", value_parser = FileModeArgParser)]
pub mode: Option<NonZero<u32>>,
#[arg(long = "no-unpack", action)]
pub do_not_unpack: bool,
#[arg(long = "only-arch", value_name = "ARCH")]
pub target_arch: Option<String>,
#[arg(value_name = "URL")]
pub url: String,
#[arg(value_name = "TARGET_PATH")]
pub target: Option<String>,
}
#[derive(Serialize, Deserialize, Debug, Clone)]
#[serde(tag = "type", rename_all = "lowercase")]
pub enum Artifact {
Tar(Tar),
Dir(Tree),
File(File),
Text(Text),
}
#[derive(Serialize, Deserialize, Debug, Clone)]
pub struct Tar {
#[serde(skip_serializing_if = "Option::is_none")]
arch: Option<String>,
#[serde(skip, default)]
uri: String,
hash: Hash,
size: u64,
#[serde(skip_serializing_if = "Option::is_none")]
target: Option<String>,
}
pub struct TarReader<'a, R: AsyncRead + Send + 'a, FS: ?Sized> {
target: Option<String>,
inner: tar::TarReader<'a, R>,
_marker: std::marker::PhantomData<fn(&FS)>,
}
#[derive(Serialize, Deserialize, Debug, Clone)]
pub struct Tree {
#[serde(skip_serializing_if = "Option::is_none")]
arch: Option<String>,
#[serde(skip, default)]
path: String,
hash: Hash,
size: u64,
#[serde(skip_serializing_if = "Option::is_none")]
target: Option<String>,
}
pub struct TreeReader<FS: ?Sized> {
target: Option<String>,
dir: OwnedFd,
hash: Hash,
source: PathBuf,
_marker: std::marker::PhantomData<fn(&FS)>,
}
#[derive(Serialize, Deserialize, Debug, Clone)]
pub struct File {
#[serde(skip_serializing_if = "Option::is_none")]
arch: Option<String>,
#[serde(skip, default)]
uri: String,
hash: Hash,
size: u64,
target: String,
#[serde(skip_serializing_if = "Option::is_none")]
mode: Option<NonZero<u32>>,
#[serde(skip_serializing_if = "Option::is_none")]
unpack: Option<bool>,
}
pub struct FileReader<'a, R: AsyncRead + Send + 'a, FS: ?Sized> {
target: String,
inner: R,
mode: Option<NonZero<u32>>,
uri: String,
unpack: bool,
size: Option<NonZero<usize>>,
_marker: std::marker::PhantomData<&'a ()>,
_marker_fs: std::marker::PhantomData<fn(&FS)>,
}
#[derive(Serialize, Deserialize, Debug, Clone)]
pub struct Text {
#[serde(skip_serializing_if = "Option::is_none")]
arch: Option<String>,
#[serde(skip, default)]
uri: String,
text: String,
target: String,
#[serde(skip_serializing_if = "Option::is_none")]
mode: Option<NonZero<u32>>,
}
pub struct TextReader<FS: ?Sized> {
target: String,
bytes: std::sync::Arc<[u8]>,
mode: Option<NonZero<u32>>,
uri: String,
_marker: std::marker::PhantomData<fn(&FS)>,
}
impl Artifact {
pub(crate) fn text(
uri: String,
target: String,
text: String,
mode: Option<NonZero<u32>>,
arch: Option<String>,
) -> Self {
Artifact::Text(Text {
arch,
uri,
text,
target,
mode,
})
}
pub(crate) fn as_text(&self) -> Option<&Text> {
match self {
Artifact::Text(text) => Some(text),
_ => None,
}
}
pub(crate) async fn new<C>(
artifact: &ArtifactArg,
local_base: Option<&Path>,
cache: &C,
) -> io::Result<Self>
where
C: ContentProvider,
{
let uri = artifact.url.clone();
let target = artifact.target.clone();
let unpack = if artifact.do_not_unpack {
Some(false)
} else {
None
};
let arch = artifact.target_arch.clone();
if let Some(target) = target.as_deref() {
if target.is_empty() {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
"target path cannot be empty",
));
}
}
if is_url(&uri) {
let mut artifact = if unpack.unwrap_or(true) && is_tar_ext(uri.as_bytes()) {
Artifact::Tar(Tar {
arch,
uri,
hash: Hash::default(),
size: 0,
target,
})
} else {
let file_target = target.as_deref().ok_or_else(|| {
io::Error::new(
io::ErrorKind::InvalidInput,
"target must be specified for a file",
)
})?;
Artifact::File(File {
arch,
uri,
hash: Hash::default(),
size: 0,
target: file_target.to_string(),
mode: artifact.mode,
unpack,
})
};
cache.ensure_artifact(&mut artifact, None).await?;
Ok(artifact)
} else {
let path = local_base.ok_or_else(|| {
io::Error::other(format!(
"missing local base path for artifact {}",
artifact.url
))
})?;
let mut artifact = if smol::fs::symlink_metadata(&path).await?.is_dir() {
Artifact::Dir(Tree {
arch,
path: uri,
hash: Hash::default(),
size: 0,
target,
})
} else if unpack.unwrap_or(true) && is_tar_ext(path.as_os_str().as_encoded_bytes()) {
Artifact::Tar(Tar {
arch,
uri,
hash: Hash::default(),
size: 0,
target,
})
} else {
let file_target = target.as_deref().ok_or_else(|| {
io::Error::new(
io::ErrorKind::InvalidInput,
"target must be specified for a file",
)
})?;
Artifact::File(File {
arch,
uri,
hash: Hash::default(),
size: 0,
target: file_target.to_string(),
mode: artifact.mode,
unpack,
})
};
cache.ensure_artifact(&mut artifact, Some(path)).await?;
Ok(artifact)
}
}
pub fn is_remote(&self) -> bool {
match &self {
Artifact::Tar(inner) => is_url(&inner.uri),
Artifact::File(inner) => is_url(&inner.uri),
Artifact::Dir(_) => false,
Artifact::Text(_) => false,
}
}
pub fn is_local(&self) -> bool {
!self.is_remote()
}
pub async fn hash_remote<R: AsyncRead + Send + 'static>(
&mut self,
r: R,
) -> io::Result<(Hash, u64)> {
match self {
Artifact::Dir(_) => Err(io::Error::other(
"directory artifacts do not support remote readers",
)),
Artifact::Tar(inner) => inner.hash_remote(r).await,
Artifact::File(inner) => inner.hash_remote(r).await,
Artifact::Text(_) => Err(io::Error::other(
"inline text artifacts do not support remote readers",
)),
}
}
pub async fn hash_stage_remote<
R: AsyncRead + Send + 'static,
FS: StagingFileSystem + ?Sized + 'static,
>(
&mut self,
r: R,
fs: &FS,
) -> io::Result<(Hash, u64)> {
match self {
Artifact::Dir(_) => Err(io::Error::other(
"directory artifacts do not support remote readers",
)),
Artifact::Tar(inner) => inner.hash_stage_remote(r, fs).await,
Artifact::File(inner) => inner.hash_stage_remote(r, fs).await,
Artifact::Text(_) => Err(io::Error::other(
"inline text artifacts do not support remote readers",
)),
}
}
pub fn remote<R: AsyncRead + Send + 'static, FS: StagingFileSystem + ?Sized + 'static>(
&self,
r: R,
) -> io::Result<Box<dyn Stage<Target = FS, Output = ()> + Send>> {
match &self {
Artifact::Dir(_) => Err(io::Error::other(
"directory artifacts do not support remote readers",
)),
Artifact::Tar(inner) => inner.remote(r),
Artifact::File(inner) => inner.remote(r),
Artifact::Text(_) => Err(io::Error::other(
"inline text artifacts do not support remote readers",
)),
}
}
pub async fn hash_stage_local<P: AsRef<Path>, FS: StagingFileSystem + ?Sized>(
&mut self,
path: P,
fs: &FS,
) -> io::Result<(Hash, u64)> {
match self {
Artifact::Dir(inner) => inner.hash_stage_local(path, fs).await,
Artifact::Tar(inner) => inner.hash_stage_local(path, fs).await,
Artifact::File(inner) => inner.hash_stage_local(path, fs).await,
Artifact::Text(inner) => inner.hash_stage_local(path, fs).await,
}
}
pub async fn hash_local<P: AsRef<Path>>(&mut self, path: P) -> io::Result<(Hash, u64)> {
match self {
Artifact::Dir(inner) => inner.hash_local(path).await,
Artifact::Tar(inner) => inner.hash_local(path).await,
Artifact::File(inner) => inner.hash_local(path).await,
Artifact::Text(inner) => inner.hash_local(path).await,
}
}
pub async fn local<P: AsRef<Path>, FS: StagingFileSystem + ?Sized + 'static>(
&self,
path: P,
) -> io::Result<Box<dyn Stage<Target = FS, Output = ()> + Send>> {
match self {
Artifact::Dir(inner) => inner.local(path).await,
Artifact::Tar(inner) => inner.local(path).await,
Artifact::File(inner) => inner.local(path).await,
Artifact::Text(inner) => inner.local(path).await,
}
}
pub fn hash(&self) -> Hash {
match self {
Artifact::Tar(inner) => inner.hash.clone(),
Artifact::Dir(inner) => inner.hash.clone(),
Artifact::File(inner) => inner.hash.clone(),
Artifact::Text(inner) => inner.hash(),
}
}
pub fn size(&self) -> u64 {
match self {
Artifact::Tar(inner) => inner.size,
Artifact::Dir(inner) => inner.size,
Artifact::File(inner) => inner.size,
Artifact::Text(inner) => inner.size(),
}
}
pub fn uri(&self) -> &str {
match self {
Artifact::Tar(inner) => &inner.uri,
Artifact::Dir(inner) => &inner.path,
Artifact::File(inner) => &inner.uri,
Artifact::Text(inner) => &inner.uri,
}
}
pub fn target(&self) -> Option<&str> {
match self {
Artifact::Tar(inner) => inner.target.as_deref(),
Artifact::Dir(inner) => inner.target.as_deref(),
Artifact::File(inner) => Some(&inner.target),
Artifact::Text(inner) => Some(&inner.target),
}
}
pub fn update_spec_hash<H: HashAlgo>(&self, hasher: &mut H) {
match self {
Artifact::Tar(inner) => inner.update_spec_hash(hasher),
Artifact::Dir(inner) => inner.update_spec_hash(hasher),
Artifact::File(inner) => inner.update_spec_hash(hasher),
Artifact::Text(inner) => inner.update_spec_hash(hasher),
}
}
pub fn arch(&self) -> Option<&str> {
match self {
Artifact::Tar(inner) => inner.arch.as_deref(),
Artifact::Dir(inner) => inner.arch.as_deref(),
Artifact::File(inner) => inner.arch.as_deref(),
Artifact::Text(inner) => inner.arch.as_deref(),
}
}
pub(crate) fn with_uri<S: Into<String>>(mut self, uri: S) -> Self {
match &mut self {
Artifact::Tar(inner) => inner.uri = uri.into(),
Artifact::Dir(inner) => inner.path = uri.into(),
Artifact::File(inner) => inner.uri = uri.into(),
Artifact::Text(inner) => inner.uri = uri.into(),
}
self
}
pub(crate) fn toml_table(&self) -> toml_edit::Table {
let mut table = toml_edit::ser::to_document(self)
.expect("failed to serialize table")
.into_table();
if let Some(mode) = match self {
Artifact::File(File {
mode: Some(mode), ..
}) => Some(mode),
Artifact::Text(Text {
mode: Some(mode), ..
}) => Some(mode),
_ => None,
} {
*table.get_mut("mode").expect("mode field") = format!("0o{:03o}", mode.get())
.parse::<toml_edit::Item>()
.expect("parsed item");
}
table
}
}
impl Tar {
fn update_spec_hash<H: HashAlgo>(&self, hasher: &mut H) {
"tar".hash_into(hasher);
if let Some(arch) = self.arch.as_deref() {
true.hash_into(hasher);
arch.hash_into(hasher);
} else {
false.hash_into(hasher);
}
self.uri.hash_into(hasher);
if let Some(target) = self.target.as_deref() {
true.hash_into(hasher);
target.hash_into(hasher);
} else {
false.hash_into(hasher);
}
self.hash.as_bytes().hash_into(hasher);
}
pub async fn local<P: AsRef<Path>, FS: StagingFileSystem + ?Sized + 'static>(
&self,
path: P,
) -> io::Result<Box<dyn Stage<Target = FS, Output = ()> + Send>> {
let r = smol::fs::File::open(path.as_ref()).await.map_err(|err| {
io::Error::other(format!(
"failed to open local tar file {}: {}",
path.as_ref().display(),
err
))
})?;
Ok(Box::new(tar_reader(&self.uri, r).map(|inner| TarReader {
inner,
target: self.target.clone(),
_marker: std::marker::PhantomData,
})?)
as Box<dyn Stage<Target = FS, Output = ()> + Send>)
}
async fn hash_stage_local<P: AsRef<Path>, FS: StagingFileSystem + ?Sized>(
&mut self,
path: P,
fs: &FS,
) -> io::Result<(Hash, u64)> {
let file = smol::fs::File::open(path.as_ref()).await.map_err(|err| {
io::Error::other(format!(
"failed to open local tar file {}: {}",
path.as_ref().display(),
err
))
})?;
self.hash_stage_remote(file, fs).await
}
async fn hash_local<P: AsRef<Path>>(&mut self, path: P) -> io::Result<(Hash, u64)> {
let file = smol::fs::File::open(path.as_ref()).await.map_err(|err| {
io::Error::other(format!(
"failed to open local tar file {}: {}",
path.as_ref().display(),
err
))
})?;
self.hash_remote(file).await
}
fn remote<R: AsyncRead + Send + 'static, FS: StagingFileSystem + ?Sized + 'static>(
&self,
r: R,
) -> io::Result<Box<dyn Stage<Target = FS, Output = ()> + Send>> {
Ok(Box::new(tar_reader(&self.uri, r).map(|inner| TarReader {
inner,
target: self.target.clone(),
_marker: std::marker::PhantomData,
})?)
as Box<dyn Stage<Target = FS, Output = ()> + Send>)
}
async fn hash_stage_remote<R: AsyncRead + Send + 'static, FS: StagingFileSystem + ?Sized>(
&mut self,
r: R,
fs: &FS,
) -> io::Result<(Hash, u64)> {
let mut reader = pin!(HashingReader::<blake3::Hasher, _>::new(r));
TarReader {
inner: tar_reader(&self.uri, &mut reader)?,
target: self.target.clone(),
_marker: std::marker::PhantomData,
}
.stage(fs)
.await?;
self.size = reader.as_mut().size();
self.hash = reader.as_mut().hash();
Ok((self.hash.clone(), self.size))
}
async fn hash_remote<R: AsyncRead + Send + 'static>(
&mut self,
r: R,
) -> io::Result<(Hash, u64)> {
let mut reader = pin!(HashingReader::<blake3::Hasher, _>::new(r));
TarReader {
inner: tar_reader(&self.uri, &mut reader)?,
target: self.target.clone(),
_marker: std::marker::PhantomData,
}
.stage(&FileList::new())
.await?;
self.size = reader.as_mut().size();
self.hash = reader.as_mut().hash();
Ok((self.hash.clone(), self.size))
}
}
impl<'a, R, FS> Stage for TarReader<'a, R, FS>
where
R: AsyncRead + Send + 'a,
FS: StagingFileSystem + ?Sized,
{
type Target = FS;
type Output = ();
fn stage<'b>(
&'b mut self,
fs: &'b Self::Target,
) -> Pin<Box<dyn Future<Output = io::Result<()>> + 'b>> {
Box::pin(self.extract_to(fs))
}
}
fn join_path<'a>(base: Option<&str>, path: &'a str) -> Cow<'a, Path> {
base.map_or_else(
|| Cow::Borrowed(Path::new(path)),
|base| Cow::Owned(Path::new(base).join(path)),
)
}
impl<'a, R, FS> TarReader<'a, R, FS>
where
R: AsyncRead + Send + 'a,
FS: StagingFileSystem + ?Sized,
{
async fn extract_to(&mut self, fs: &FS) -> io::Result<()> {
let mut links: Vec<tar::TarLink> = Vec::new();
let target = match self.target.as_deref() {
Some(target) => Some(stage_dir_target(target)?),
None => None,
};
let target_path = target.as_deref();
if let Some(parent) = target_path.and_then(|t| Path::new(t).parent()) {
fs.create_dir_all(parent, 0, 0, 0o755).await?;
}
while let Some(entry) = self.inner.next().await {
let entry = entry?;
match entry {
tar::TarEntry::Directory(dir) => {
validate_unpacked_path(dir.path())?;
let path = join_path(target_path, dir.path());
tracing::trace!("creating directory {}", path.display());
fs.create_dir_all(path, dir.uid(), dir.gid(), dir.mode())
.await?;
}
tar::TarEntry::File(mut file) => {
validate_unpacked_path(file.path())?;
let path = target_path.map_or_else(
|| Path::new(file.path()).to_path_buf(),
|p| Path::new(p).join(file.path()),
);
if let Some(parent) = path.parent() {
fs.create_dir_all(parent, 0, 0, 0o755).await?;
}
let size = file.size() as usize;
let uid = file.uid();
let gid = file.gid();
let mode = file.mode();
tracing::trace!("extracting {}", path.display());
fs.create_file(&mut file, uid, gid, mode, Some(size))
.await
.map_err(|err| {
io::Error::other(format!(
"error creating file {}: {}",
path.display(),
err
))
})?
.persist(&path)
.await?;
}
tar::TarEntry::Symlink(link) => {
validate_unpacked_path(link.path())?;
let uid = link.uid();
let gid = link.gid();
let path = join_path(target_path, link.path());
let link = Path::new(link.link());
let link = target_path.map_or_else(
|| Cow::Borrowed(link),
|p| {
if link.is_absolute() {
Cow::Owned(Path::new(p).join(link))
} else {
Cow::Borrowed(link)
}
},
);
fs.symlink(link, path, uid, gid).await?;
}
tar::TarEntry::Link(link) => {
links.push(link);
}
_ => {
return Err(io::Error::other(format!(
"unsupported tar entry in artifact {} {:?}",
&self.target.as_deref().unwrap_or("<no-target>"),
&entry,
)));
}
}
}
for link in links.drain(..) {
validate_unpacked_path(link.path())?;
validate_unpacked_path(link.link())?;
let path = join_path(target_path, link.path());
let link = join_path(target_path, link.link());
fs.hardlink(link, path).await?;
}
Ok(())
}
}
impl Tree {
fn update_spec_hash<H: HashAlgo>(&self, hasher: &mut H) {
"dir".hash_into(hasher);
if let Some(arch) = self.arch.as_deref() {
true.hash_into(hasher);
arch.hash_into(hasher);
} else {
false.hash_into(hasher);
}
self.path.hash_into(hasher);
if let Some(target) = self.target.as_deref() {
true.hash_into(hasher);
target.hash_into(hasher);
} else {
false.hash_into(hasher);
}
self.hash.as_bytes().hash_into(hasher);
}
async fn hash_stage_local<P: AsRef<Path>, FS: StagingFileSystem + ?Sized>(
&mut self,
path: P,
fs: &FS,
) -> io::Result<(Hash, u64)> {
let fd = openat(
CWD,
path.as_ref(),
OFlags::DIRECTORY | OFlags::RDONLY | OFlags::CLOEXEC | OFlags::NOFOLLOW,
Mode::empty(),
)
.map_err(|err| {
io::Error::other(format!(
"failed to open local artifact directory {}: {}",
path.as_ref().display(),
err
))
})?;
let target = match self.target.as_deref() {
Some(target) => Some(stage_dir_target(target)?),
None => None,
};
let (hash, size) =
tree::copy_hash_dir_inner::<blake3::Hasher, _, _>(&fd, fs, target.as_deref())
.await
.map(|(hash, size)| (hash.hash(), size))
.map_err(|err| {
io::Error::other(format!(
"failed to stage local artifact directory {}: {}",
path.as_ref().display(),
err
))
})?;
self.hash = hash.clone();
self.size = size;
Ok((hash, size))
}
async fn hash_local<P: AsRef<Path>>(&mut self, path: P) -> io::Result<(Hash, u64)> {
let fd = openat(
CWD,
path.as_ref(),
OFlags::DIRECTORY | OFlags::RDONLY | OFlags::CLOEXEC | OFlags::NOFOLLOW,
Mode::empty(),
)
.map_err(|err| {
io::Error::other(format!(
"failed to open local artifact directory {}: {}",
path.as_ref().display(),
err
))
})?;
let (hash, size) = tree::hash_dir_inner::<blake3::Hasher>(fd)
.await
.map(|(hash, size)| (hash.hash(), size))
.map_err(|err| {
io::Error::other(format!(
"failed to hash local artifact directory {}: {}",
path.as_ref().display(),
err
))
})?;
self.hash = hash.clone();
self.size = size;
Ok((hash, size))
}
async fn local<P: AsRef<Path>, FS: StagingFileSystem + ?Sized + 'static>(
&self,
path: P,
) -> io::Result<Box<dyn Stage<Target = FS, Output = ()> + Send>> {
let fd = openat(
CWD,
path.as_ref(),
OFlags::DIRECTORY | OFlags::RDONLY | OFlags::CLOEXEC | OFlags::NOFOLLOW,
Mode::empty(),
)
.map_err(|err| {
io::Error::other(format!(
"failed to open local artifact directory {}: {}",
path.as_ref().display(),
err
))
})?;
Ok(Box::new(TreeReader {
dir: fd,
target: self.target.clone(),
hash: self.hash.clone(),
source: path.as_ref().to_path_buf(),
_marker: std::marker::PhantomData,
})
as Box<dyn Stage<Target = FS, Output = ()> + Send>)
}
}
pub(crate) async fn hash_directory<P: AsRef<Path>>(path: P, hash: &str) -> io::Result<(Hash, u64)> {
let fd = openat(
CWD,
path.as_ref(),
OFlags::DIRECTORY | OFlags::RDONLY | OFlags::CLOEXEC | OFlags::NOFOLLOW,
Mode::empty(),
)
.map_err(|err| {
io::Error::other(format!(
"failed to open directory {}: {}",
path.as_ref().display(),
err
))
})?;
tree::hash_dir(fd, hash).await
}
impl<FS> Stage for TreeReader<FS>
where
FS: StagingFileSystem + ?Sized,
{
type Target = FS;
type Output = ();
fn stage<'b>(
&'b mut self,
fs: &'b Self::Target,
) -> Pin<Box<dyn Future<Output = io::Result<()>> + 'b>> {
Box::pin(async move {
let source = self.source.display().to_string();
let target = match self.target.as_deref() {
Some(target) => Some(stage_dir_target(target)?),
None => None,
};
if let Some(parent) = target.as_deref().and_then(|t| Path::new(t).parent()) {
fs.create_dir_all(parent, 0, 0, 0o755)
.await
.map_err(|err| {
io::Error::other(format!(
"failed to create parent directory {} for artifact {}: {}",
parent.display(),
source.as_str(),
err
))
})?;
}
let (hash, _) = tree::copy_hash_dir(&self.dir, fs, target.as_deref(), self.hash.name())
.await
.map_err(|err| {
io::Error::other(format!(
"failed to stage local directory artifact {}: {}",
source.as_str(),
err
))
})?;
if hash != self.hash {
Err(io::Error::other(format!(
"hash mismatch after copying local tree {}",
source.as_str()
)))
} else {
Ok(())
}
})
}
}
impl File {
fn update_spec_hash<H: HashAlgo>(&self, hasher: &mut H) {
"file".hash_into(hasher);
if let Some(arch) = self.arch.as_deref() {
true.hash_into(hasher);
arch.hash_into(hasher);
} else {
false.hash_into(hasher);
}
self.uri.hash_into(hasher);
self.target.hash_into(hasher);
let mode = self.mode.map_or(0o644, |m| m.get() & 0o7777);
mode.hash_into(hasher);
self.unpack.unwrap_or(true).hash_into(hasher);
self.hash.as_bytes().hash_into(hasher);
}
async fn local<P: AsRef<Path>, FS: StagingFileSystem + ?Sized + 'static>(
&self,
path: P,
) -> io::Result<Box<dyn Stage<Target = FS, Output = ()> + Send>> {
let r = smol::fs::File::open(path.as_ref()).await.map_err(|err| {
io::Error::other(format!(
"failed to open local artifact file {}: {}",
path.as_ref().display(),
err
))
})?;
Ok(Box::new(FileReader {
inner: self.hash.reader(self.size, r),
target: self.target.clone(),
mode: self.mode,
uri: self.uri.clone(),
unpack: self.unpack.unwrap_or(true),
size: Some(NonZero::new(self.size as usize).unwrap()),
_marker: std::marker::PhantomData,
_marker_fs: std::marker::PhantomData,
})
as Box<dyn Stage<Target = FS, Output = ()> + Send>)
}
async fn hash_stage_local<P: AsRef<Path>, FS: StagingFileSystem + ?Sized>(
&mut self,
path: P,
fs: &FS,
) -> io::Result<(Hash, u64)> {
let fd = openat(
CWD,
path.as_ref(),
OFlags::RDONLY | OFlags::CLOEXEC | OFlags::NOFOLLOW,
Mode::empty(),
)
.map_err(|err| {
io::Error::other(format!(
"failed to open local artifact file {}: {}",
path.as_ref().display(),
err
))
})?;
let stat = fstat(&fd).map_err(|err| {
io::Error::other(format!(
"failed to stat local artifact file {}: {}",
path.as_ref().display(),
err
))
})?;
if !FileType::from_raw_mode(stat.st_mode).is_file() {
return Err(io::Error::other(format!(
"artefact file {} ({}) is not a regular file",
&self.uri,
path.as_ref().display()
)));
}
let st_mode = stat.st_mode & 0o7777;
if st_mode == 0 {
return Err(io::Error::other(format!(
"file {} has invalid permissions set {:0o}",
&self.uri, st_mode,
)));
}
if self.mode.is_none() && st_mode != 0o644 {
self.mode = Some(NonZero::new(st_mode as u32).unwrap());
}
self.hash_stage_remote(smol::fs::File::from(fd), fs).await
}
async fn hash_local<P: AsRef<Path>>(&mut self, path: P) -> io::Result<(Hash, u64)> {
let fd = openat(
CWD,
path.as_ref(),
OFlags::RDONLY | OFlags::CLOEXEC | OFlags::NOFOLLOW,
Mode::empty(),
)
.map_err(|err| {
io::Error::other(format!(
"failed to open local artifact file {}: {}",
path.as_ref().display(),
err
))
})?;
let stat = fstat(&fd).map_err(|err| {
io::Error::other(format!(
"failed to stat local artifact file {}: {}",
path.as_ref().display(),
err
))
})?;
if !FileType::from_raw_mode(stat.st_mode).is_file() {
return Err(io::Error::other(format!(
"artefact file {} ({}) is not a regular file",
&self.uri,
path.as_ref().display()
)));
}
let st_mode = stat.st_mode & 0o7777;
if st_mode == 0 {
return Err(io::Error::other(format!(
"file {} has invalid permissions set {:0o}",
&self.uri, st_mode,
)));
}
if self.mode.is_none() && st_mode != 0o644 {
self.mode = Some(NonZero::new(st_mode as u32).unwrap());
}
let (hash, size) = if stat.st_size < 65_536 {
let mut reader = HashingReader::<blake3::Hasher, _>::new(smol::fs::File::from(fd));
smol::io::copy(&mut reader, smol::io::sink())
.await
.map_err(|err| {
io::Error::other(format!(
"failed to hash local artifact file {}: {}",
path.as_ref().display(),
err
))
})?;
reader.into_hash_and_size()
} else {
let hash = smol::unblock(move || {
let map = unsafe { memmap2::Mmap::map(fd.as_raw_fd()) }?;
map.advise(memmap2::Advice::Sequential)?;
let mut hasher = blake3::Hasher::new();
hasher.update(&map[..]);
Ok::<_, io::Error>(hasher.into_hash())
})
.await
.map_err(|err| {
io::Error::other(format!(
"failed to hash local artifact file {}: {}",
path.as_ref().display(),
err
))
})?;
(hash, stat.st_size as u64)
};
self.hash = hash.clone();
self.size = size;
Ok((hash, size))
}
async fn hash_stage_remote<R: AsyncRead + Send + 'static, FS: StagingFileSystem + ?Sized>(
&mut self,
r: R,
fs: &FS,
) -> io::Result<(Hash, u64)> {
let st_mode = self.mode.map_or(0o644, |m| m.get() & 0o7777);
let uri = self.uri.as_str();
let target = stage_file_target(self.target.as_str(), uri)?;
let mut reader = pin!(HashingReader::<blake3::Hasher, _>::new(r));
if let Some(parent) = Path::new(target.as_ref()).parent() {
fs.create_dir_all(parent, 0, 0, 0o755)
.await
.map_err(|err| {
io::Error::other(format!(
"failed to create parent directory {} for artifact {}: {}",
parent.display(),
uri,
err
))
})?;
}
if self.unpack.unwrap_or(true) && is_comp_ext(&self.uri) {
fs.create_file(comp_reader(&self.uri, &mut reader), 0, 0, st_mode, None)
.await
.map_err(|err| {
io::Error::other(format!(
"failed to create target file {} for artifact {}: {}",
target, uri, err
))
})?
} else {
fs.create_file(&mut reader, 0, 0, st_mode, None)
.await
.map_err(|err| {
io::Error::other(format!(
"failed to create target file {} for artifact {}: {}",
target, uri, err
))
})?
}
.persist(target.as_ref())
.await
.map_err(|err| {
io::Error::other(format!(
"failed to persist target file {} for artifact {}: {}",
target, uri, err
))
})?;
self.size = reader.as_mut().size();
self.hash = reader.as_mut().hash();
Ok((self.hash.clone(), self.size))
}
async fn hash_remote<R: AsyncRead + Send + 'static>(
&mut self,
r: R,
) -> io::Result<(Hash, u64)> {
let mut reader = pin!(HashingReader::<blake3::Hasher, _>::new(r));
smol::io::copy(&mut reader, smol::io::sink()).await?;
self.size = reader.as_mut().size();
self.hash = reader.as_mut().hash();
Ok((self.hash.clone(), self.size))
}
fn remote<R: AsyncRead + Send + 'static, FS: StagingFileSystem + ?Sized + 'static>(
&self,
r: R,
) -> io::Result<Box<dyn Stage<Target = FS, Output = ()> + Send>> {
Ok(Box::new(FileReader {
inner: self.hash.reader(self.size, r),
target: self.target.clone(),
mode: self.mode,
uri: self.uri.clone(),
unpack: self.unpack.unwrap_or(true),
size: Some(NonZero::new(self.size as usize).unwrap()),
_marker: std::marker::PhantomData,
_marker_fs: std::marker::PhantomData,
})
as Box<dyn Stage<Target = FS, Output = ()> + Send>)
}
}
impl Text {
pub(crate) fn target(&self) -> &str {
&self.target
}
pub(crate) fn text(&self) -> &str {
&self.text
}
pub(crate) fn mode(&self) -> Option<NonZero<u32>> {
self.mode
}
pub(crate) fn arch(&self) -> Option<&str> {
self.arch.as_deref()
}
fn update_spec_hash<H: HashAlgo>(&self, hasher: &mut H) {
"text".hash_into(hasher);
if let Some(arch) = self.arch.as_deref() {
true.hash_into(hasher);
arch.hash_into(hasher);
} else {
false.hash_into(hasher);
}
self.target.hash_into(hasher);
let mode = self.mode.map_or(0o644, |m| m.get() & 0o7777);
mode.hash_into(hasher);
let mut text_hash = blake3::Hasher::new();
text_hash.update(self.text.as_bytes());
let digest = text_hash.finalize();
digest.as_slice().hash_into(hasher);
}
fn size(&self) -> u64 {
self.text.len() as u64
}
fn hash(&self) -> Hash {
let mut hasher = blake3::Hasher::new();
hasher.update(self.text.as_bytes());
hasher.into_hash()
}
async fn local<P: AsRef<Path>, FS: StagingFileSystem + ?Sized + 'static>(
&self,
_path: P,
) -> io::Result<Box<dyn Stage<Target = FS, Output = ()> + Send>> {
let bytes: std::sync::Arc<[u8]> = self.text.as_bytes().to_vec().into();
Ok(Box::new(TextReader {
target: self.target.clone(),
bytes,
mode: self.mode,
uri: self.uri.clone(),
_marker: std::marker::PhantomData,
})
as Box<dyn Stage<Target = FS, Output = ()> + Send>)
}
async fn hash_stage_local<P: AsRef<Path>, FS: StagingFileSystem + ?Sized>(
&mut self,
_path: P,
_fs: &FS,
) -> io::Result<(Hash, u64)> {
let mut hasher = blake3::Hasher::new();
let bytes = self.text.as_bytes();
hasher.update(bytes);
let hash = hasher.into_hash();
Ok((hash, bytes.len() as u64))
}
async fn hash_local<P: AsRef<Path>>(&mut self, _path: P) -> io::Result<(Hash, u64)> {
let mut hasher = blake3::Hasher::new();
let bytes = self.text.as_bytes();
hasher.update(bytes);
let hash = hasher.into_hash();
Ok((hash, bytes.len() as u64))
}
}
impl<FS> Stage for TextReader<FS>
where
FS: StagingFileSystem + ?Sized,
{
type Target = FS;
type Output = ();
fn stage<'b>(
&'b mut self,
fs: &'b Self::Target,
) -> Pin<Box<dyn Future<Output = io::Result<()>> + 'b>> {
Box::pin(async move {
let uri = self.uri.as_str();
let target = stage_text_target(self.target.as_str(), uri)?;
if let Some(parent) = Path::new(&target).parent() {
fs.create_dir_all(parent, 0, 0, 0o755)
.await
.map_err(|err| {
io::Error::other(format!(
"failed to create parent directory {} for artifact {}: {}",
parent.display(),
uri,
err
))
})?;
}
fs.create_file_from_bytes(
self.bytes.as_ref(),
0,
0,
self.mode.map_or(0o644, |m| m.get() & 0o7777),
)
.await
.map_err(|err| {
io::Error::other(format!(
"failed to create target file {} for artifact {}: {}",
target, uri, err
))
})?
.persist(&target)
.await
.map_err(|err| {
io::Error::other(format!(
"failed to persist target file {} for artifact {}: {}",
target, uri, err
))
})
})
}
}
impl<'a, R, FS> Stage for FileReader<'a, R, FS>
where
R: AsyncRead + Send + Unpin + 'a,
FS: StagingFileSystem + ?Sized,
{
type Target = FS;
type Output = ();
fn stage<'b>(
&'b mut self,
fs: &'b Self::Target,
) -> Pin<Box<dyn Future<Output = io::Result<()>> + 'b>> {
Box::pin(async move {
let uri = self.uri.as_str();
let target = stage_file_target(self.target.as_str(), uri)?;
if let Some(parent) = Path::new(target.as_ref()).parent() {
fs.create_dir_all(parent, 0, 0, 0o755)
.await
.map_err(|err| {
io::Error::other(format!(
"failed to create parent directory {} for artifact {}: {}",
parent.display(),
uri,
err
))
})?;
}
if self.unpack && is_comp_ext(self.uri.as_bytes()) {
fs.create_file(
comp_reader(&self.uri, &mut self.inner),
0,
0,
self.mode.map_or(0o644, |m| m.get() & 0o7777),
None,
)
.await
.map_err(|err| {
io::Error::other(format!(
"failed to create target file {} for artifact {}: {}",
target, uri, err
))
})?
.persist(target.as_ref())
.await
.map_err(|err| {
io::Error::other(format!(
"failed to persist target file {} for artifact {}: {}",
target, uri, err
))
})
} else {
fs.create_file(
&mut self.inner,
0,
0,
self.mode.map_or(0o644, |m| m.get() & 0o7777),
self.size.map(|s| s.get()),
)
.await
.map_err(|err| {
io::Error::other(format!(
"failed to create target file {} for artifact {}: {}",
target, uri, err
))
})?
.persist(target.as_ref())
.await
.map_err(|err| {
io::Error::other(format!(
"failed to persist target file {} for artifact {}: {}",
target, uri, err
))
})
}
})
}
}
fn mode(mode: u32) -> u32 {
let ft = mode & libc::S_IFMT;
let perm = mode & 0o7777;
ft | perm
}
mod tree {
use {
super::mode,
crate::{
hash::{Hash, HashAlgo, HashingReader, InnerHash},
StagingFile, StagingFileSystem,
},
async_channel as chan,
futures::stream::{self, StreamExt, TryStreamExt},
itertools::Itertools,
rustix::{
fd::{AsFd, OwnedFd},
fs::{fstat, openat2, readlinkat, Dir, FileType, Mode, OFlags, ResolveFlags, Stat},
io::Errno,
},
std::{
ffi::{CString, OsStr, OsString},
io,
os::{fd::AsRawFd, unix::ffi::OsStrExt},
path::{Path, PathBuf},
str::FromStr,
},
};
enum Object {
Directory {
path: PathBuf,
stat: Stat,
},
Symlink {
path: PathBuf,
target: CString,
},
RegularFile {
path: PathBuf,
stat: Stat,
fd: OwnedFd,
},
}
struct SmallBlockingPool {
tx: chan::Sender<Box<dyn FnOnce() + Send + 'static>>,
}
impl SmallBlockingPool {
fn new(threads: usize, name: &str) -> Self {
let (tx, rx) = chan::unbounded::<Box<dyn FnOnce() + Send>>();
for i in 0..threads {
let rx = rx.clone();
let name = format!("{name}-{i}");
std::thread::Builder::new()
.name(name)
.spawn(move || {
while let Ok(job) = rx.recv_blocking() {
job();
}
})
.expect("spawn worker");
}
Self { tx }
}
async fn spawn<F, R>(&self, f: F) -> R
where
F: FnOnce() -> R + Send + 'static,
R: Send + 'static,
{
let (rtx, rrx) = chan::bounded(1);
let job = Box::new(move || {
let res = f();
let _ = rtx.send_blocking(res);
});
self.tx.send_blocking(job).expect("pool closed");
rrx.recv().await.expect("worker dropped")
}
}
pub(super) async fn copy_hash_dir<Fd: AsFd, FS: StagingFileSystem + ?Sized>(
fd: Fd,
fs: &FS,
target_path: Option<&str>,
hash: &str,
) -> io::Result<(Hash, u64)> {
match hash {
md5::Md5::NAME => copy_hash_dir_inner::<md5::Md5, _, _>(fd, fs, target_path)
.await
.map(|(hash, size)| (hash.hash(), size)),
sha1::Sha1::NAME => copy_hash_dir_inner::<sha1::Sha1, _, _>(fd, fs, target_path)
.await
.map(|(hash, size)| (hash.hash(), size)),
sha2::Sha256::NAME => copy_hash_dir_inner::<sha2::Sha256, _, _>(fd, fs, target_path)
.await
.map(|(hash, size)| (hash.hash(), size)),
sha2::Sha512::NAME => copy_hash_dir_inner::<sha2::Sha512, _, _>(fd, fs, target_path)
.await
.map(|(hash, size)| (hash.hash(), size)),
blake3::Hasher::NAME => {
copy_hash_dir_inner::<blake3::Hasher, _, _>(fd, fs, target_path)
.await
.map(|(hash, size)| (hash.hash(), size))
}
_ => io::Result::Err(io::Error::other(format!(
"unsupported hash algorithm for directory artifact: {hash}"
))),
}
}
pub(super) async fn copy_hash_dir_inner<
H: HashAlgo,
Fd: AsFd,
FS: StagingFileSystem + ?Sized,
>(
fd: Fd,
fs: &FS,
target_path: Option<&str>,
) -> io::Result<(InnerHash<H>, u64)> {
process(fd, &|obj| async {
match obj {
Object::Symlink { path, target } => {
let link_target = Path::new(OsStr::from_bytes(target.as_bytes()));
let path = if let Some(target_path) = target_path {
Path::new(target_path).join(path)
} else {
path
};
fs.symlink(link_target, &path, 0, 0).await?;
Ok(None)
}
Object::Directory { stat, path } => {
let path = if let Some(target_path) = target_path {
Path::new(target_path).join(path)
} else {
path
};
fs.create_dir_all(path, 0, 0, stat.st_mode).await?;
Ok(None)
}
Object::RegularFile { fd, stat, path } => {
let path = if let Some(target_path) = target_path {
Path::new(target_path).join(path)
} else {
path
};
let mut rd = HashingReader::<H, _>::new(smol::fs::File::from(fd));
let file = fs
.create_file(&mut rd, 0, 0, stat.st_mode, Some(stat.st_size as usize))
.await?;
file.persist(path).await?;
let hash: InnerHash<H> = rd.into_hash_output().into();
Ok(Some(hash))
}
}
})
.await
}
pub(super) async fn hash_dir(fd: OwnedFd, hash: &str) -> io::Result<(Hash, u64)> {
match hash {
md5::Md5::NAME => hash_dir_inner::<md5::Md5>(fd)
.await
.map(|(hash, size)| (hash.hash(), size)),
sha1::Sha1::NAME => hash_dir_inner::<sha1::Sha1>(fd)
.await
.map(|(hash, size)| (hash.hash(), size)),
sha2::Sha256::NAME => hash_dir_inner::<sha2::Sha256>(fd)
.await
.map(|(hash, size)| (hash.hash(), size)),
sha2::Sha512::NAME => hash_dir_inner::<sha2::Sha512>(fd)
.await
.map(|(hash, size)| (hash.hash(), size)),
blake3::Hasher::NAME => hash_dir_inner::<blake3::Hasher>(fd)
.await
.map(|(hash, size)| (hash.hash(), size)),
_ => io::Result::Err(io::Error::other(format!(
"unsupported hash algorithm for directory artifact: {hash}"
))),
}
}
pub(super) async fn hash_dir_inner<H: HashAlgo + 'static>(
fd: OwnedFd,
) -> io::Result<(InnerHash<H>, u64)> {
let pool = SmallBlockingPool::new(num_cpus::get().saturating_mul(2), "hash-pool");
process(fd, &|obj| async {
match obj {
Object::Symlink { .. } => Ok(None),
Object::Directory { .. } => Ok(None),
Object::RegularFile { fd, stat, path } => {
let hash = if stat.st_size < 16 * 1024 {
let mut reader = HashingReader::<H, _>::new(smol::fs::File::from(fd));
smol::io::copy(&mut reader, smol::io::sink()).await?;
Ok(reader.into_hash_output().into())
} else {
let mut hasher = H::default();
pool.spawn(move || {
let map = unsafe { memmap2::Mmap::map(fd.as_raw_fd()) }?;
map.advise(memmap2::Advice::Sequential)?;
hasher.update(&map[..]);
Ok::<_, io::Error>(hasher.finalize_fixed().into())
})
.await
}
.map_err(|err| {
io::Error::other(format!("failed to hash file {}: {}", path.display(), err))
})?;
Ok(Some(hash))
}
}
})
.await
}
async fn process<H, Fd, F, Fut>(fd: Fd, f: &F) -> io::Result<(InnerHash<H>, u64)>
where
H: HashAlgo,
Fd: AsFd,
F: Fn(Object) -> Fut,
Fut: std::future::Future<Output = io::Result<Option<InnerHash<H>>>>,
{
process_fs_object(
PathBuf::from_str("").unwrap().as_ref(),
fd,
OsStr::from_bytes(b".").to_owned(),
f,
)
.await
}
async fn process_fs_object<H, F, Fut>(
parent: &Path,
parent_dfd: impl AsFd,
name: OsString,
f: &F,
) -> io::Result<(InnerHash<H>, u64)>
where
H: HashAlgo,
F: Fn(Object) -> Fut,
Fut: std::future::Future<Output = io::Result<Option<InnerHash<H>>>>,
{
let path = parent.join(&name);
let fd = match openat2(
parent_dfd.as_fd(),
&name,
OFlags::CLOEXEC | OFlags::NOFOLLOW,
Mode::empty(),
ResolveFlags::NO_SYMLINKS | ResolveFlags::NO_XDEV,
) {
Ok(fd) => fd,
Err(Errno::LOOP) => openat2(
parent_dfd.as_fd(),
&name,
OFlags::CLOEXEC | OFlags::PATH | OFlags::NOFOLLOW,
Mode::empty(),
ResolveFlags::NO_SYMLINKS | ResolveFlags::NO_XDEV,
)
.map_err(|err| {
io::Error::other(format!("failed to open {}: {}", path.display(), err))
})?,
Err(err) => {
return Err(io::Error::other(format!(
"failed to open {}: {}",
path.display(),
err
)))
}
};
let stat = fstat(&fd).map_err(|err| {
io::Error::other(format!("failed to stat {}: {}", path.display(), err))
})?;
let mut hasher = H::default();
hasher.update(&mode(stat.st_mode).to_le_bytes());
hasher.update(name.as_encoded_bytes());
match FileType::from_raw_mode(stat.st_mode) {
FileType::RegularFile => {
let file_hash = f(Object::RegularFile { path, stat, fd })
.await?
.expect("file hash");
hasher.update(file_hash.as_bytes());
Ok((hasher.finalize_fixed().into(), stat.st_size as u64))
}
FileType::Symlink => {
let target = readlinkat(&fd, c"", []).map_err(|err| {
io::Error::other(format!(
"failed to read symlink {}: {}",
path.display(),
err
))
})?;
hasher.update(target.as_bytes());
f(Object::Symlink { path, target }).await?;
Ok((hasher.finalize_fixed().into(), 1))
}
FileType::Directory => {
f(Object::Directory {
path: path.clone(),
stat,
})
.await?;
let (dir_hash, dir_size) = process_dir::<H, _, _>(path, fd, f).await?;
hasher.update(dir_hash.as_bytes());
Ok((hasher.finalize_fixed().into(), dir_size))
}
_ => Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!("unsupported file type at {}", path.display()),
)),
}
}
async fn process_dir<H, F, Fut>(
path: PathBuf,
fd: OwnedFd,
f: &F,
) -> io::Result<(InnerHash<H>, u64)>
where
H: HashAlgo,
F: Fn(Object) -> Fut,
Fut: std::future::Future<Output = io::Result<Option<InnerHash<H>>>>,
{
let dir = Dir::read_from(&fd).map_err(|err| {
io::Error::other(format!(
"failed to read directory {}: {}",
path.display(),
err
))
})?;
let mut entries = dir
.filter_ok(|entry| entry.file_name() != c"." && entry.file_name() != c"..")
.map_ok(|entry| OsStr::from_bytes(entry.file_name().to_bytes()).to_owned())
.collect::<Result<Vec<_>, _>>()?;
entries.sort_by(|a, b| a.as_encoded_bytes().cmp(b.as_encoded_bytes()));
let (hasher, size) = stream::iter(
entries
.into_iter()
.map(|entry| process_fs_object::<H, _, _>(&path, &fd, entry, f)),
)
.buffered(16)
.try_fold(
(H::default(), 0u64),
|(mut hasher, size), (hash, s)| async move {
hasher.update(hash.as_bytes());
Ok((hasher, size + s))
},
)
.await?;
Ok((hasher.finalize_fixed().into(), size))
}
}
#[cfg(test)]
mod tests {
use super::{
normalize_target_absolute, stage_dir_target, stage_file_target, stage_text_target,
};
#[test]
fn normalize_target_rejects_empty_and_prefixes_relative() {
let err = normalize_target_absolute("").unwrap_err();
assert_eq!(err.kind(), std::io::ErrorKind::InvalidInput);
let abs = normalize_target_absolute("/foo").unwrap();
assert_eq!(abs.as_ref(), "/foo");
let rel = normalize_target_absolute("bar").unwrap();
assert_eq!(rel.as_ref(), "/bar");
}
#[test]
fn stage_file_target_appends_filename_for_trailing_slash() {
let target = stage_file_target("/opt/target/", "dir/file.txt").expect("target");
assert_eq!(target, "/opt/target/file.txt");
}
#[test]
fn stage_file_target_prefixes_relative_paths() {
let target = stage_file_target("etc/config", "config").expect("target");
assert_eq!(target, "/etc/config");
}
#[test]
fn stage_text_target_appends_artifact_key() {
let target = stage_text_target("/etc/", "artifact-name").expect("target");
assert_eq!(target, "/etc/artifact-name");
}
#[test]
fn stage_dir_target_prefixes_relative_paths() {
let target = stage_dir_target("opt/data").expect("target");
assert_eq!(target, "/opt/data");
}
}