use crate::{Error, PackageIndex, Store, StoredFile};
use std::path::Path;
pub fn directory_content_fingerprint(dir: &Path) -> Result<String, Error> {
let entries = collect_directory_fingerprints(dir, true)?;
Ok(content_fingerprint(&entries))
}
pub fn directory_metadata_fingerprint(dir: &Path) -> Result<String, Error> {
let entries = collect_directory_fingerprints(dir, false)?;
Ok(metadata_fingerprint(&entries))
}
pub fn directory_fingerprints(dir: &Path) -> Result<(String, String), Error> {
let entries = collect_directory_fingerprints(dir, true)?;
Ok((
content_fingerprint(&entries),
metadata_fingerprint(&entries),
))
}
#[derive(Debug)]
struct DirectoryFileFingerprint {
path: String,
content_hash: Option<String>,
executable: bool,
size: u64,
mtime_secs: i64,
mtime_nanos: u32,
}
fn content_fingerprint(entries: &[DirectoryFileFingerprint]) -> String {
let mut entries: Vec<(&str, &str, bool)> = entries
.iter()
.filter_map(|entry| {
Some((
entry.path.as_str(),
entry.content_hash.as_deref()?,
entry.executable,
))
})
.collect();
entries.sort_unstable();
let mut hasher = blake3::Hasher::new();
for (path, hex_hash, executable) in entries {
hasher.update(path.as_bytes());
hasher.update(b"\0");
hasher.update(hex_hash.as_bytes());
hasher.update(if executable { b"\x01" } else { b"\x00" });
}
hasher.finalize().to_hex().to_string()
}
fn metadata_fingerprint(entries: &[DirectoryFileFingerprint]) -> String {
let mut entries: Vec<&DirectoryFileFingerprint> = entries.iter().collect();
entries.sort_unstable_by(|a, b| a.path.cmp(&b.path));
let mut hasher = blake3::Hasher::new();
for entry in entries {
hasher.update(entry.path.as_bytes());
hasher.update(b"\0");
hasher.update(&entry.size.to_le_bytes());
hasher.update(&entry.mtime_secs.to_le_bytes());
hasher.update(&entry.mtime_nanos.to_le_bytes());
hasher.update(if entry.executable { b"\x01" } else { b"\x00" });
}
hasher.finalize().to_hex().to_string()
}
fn collect_directory_fingerprints(
dir: &Path,
hash_content: bool,
) -> Result<Vec<DirectoryFileFingerprint>, Error> {
let mut entries = Vec::new();
collect_directory_fingerprints_recursive(dir, dir, hash_content, &mut entries)?;
Ok(entries)
}
fn collect_directory_fingerprints_recursive(
base: &Path,
current: &Path,
hash_content: bool,
entries: &mut Vec<DirectoryFileFingerprint>,
) -> Result<(), Error> {
let dir_entries = std::fs::read_dir(current)
.map_err(|e| Error::Tar(format!("read_dir {}: {e}", current.display())))?;
for entry in dir_entries {
let entry = entry.map_err(|e| Error::Tar(format!("read_dir entry: {e}")))?;
let file_type = entry
.file_type()
.map_err(|e| Error::Tar(format!("file_type: {e}")))?;
let name = entry.file_name();
let name = name.to_string_lossy();
if matches!(name.as_ref(), ".git" | "node_modules") {
continue;
}
let path = entry.path();
if file_type.is_dir() {
collect_directory_fingerprints_recursive(base, &path, hash_content, entries)?;
continue;
}
if !file_type.is_file() {
continue;
}
let metadata = entry
.metadata()
.map_err(|e| Error::Tar(format!("metadata {}: {e}", path.display())))?;
let content_hash = if hash_content {
let content = std::fs::read(&path)
.map_err(|e| Error::Tar(format!("read {}: {e}", path.display())))?;
Some(blake3::hash(&content).to_hex().to_string())
} else {
None
};
#[cfg(unix)]
let executable = {
use std::os::unix::fs::PermissionsExt;
metadata.permissions().mode() & 0o111 != 0
};
#[cfg(not(unix))]
let executable = false;
let rel = path
.strip_prefix(base)
.map_err(|e| Error::Tar(format!("strip_prefix: {e}")))?
.to_string_lossy()
.replace('\\', "/");
let modified = metadata
.modified()
.ok()
.and_then(|time| time.duration_since(std::time::UNIX_EPOCH).ok());
let (mtime_secs, mtime_nanos) = modified
.map(|duration| (duration.as_secs() as i64, duration.subsec_nanos()))
.unwrap_or((0, 0));
entries.push(DirectoryFileFingerprint {
path: rel,
content_hash,
executable,
size: metadata.len(),
mtime_secs,
mtime_nanos,
});
}
Ok(())
}
impl Store {
pub fn import_directory(&self, dir: &Path) -> Result<PackageIndex, Error> {
let mut index = PackageIndex::default();
self.import_directory_recursive(dir, dir, &mut index)?;
Ok(index)
}
fn import_directory_recursive(
&self,
base: &Path,
current: &Path,
index: &mut PackageIndex,
) -> Result<(), Error> {
let entries = std::fs::read_dir(current)
.map_err(|e| Error::Tar(format!("read_dir {}: {e}", current.display())))?;
for entry in entries {
let entry =
entry.map_err(|e| Error::Tar(format!("read_dir {}: {e}", current.display())))?;
let file_type = entry
.file_type()
.map_err(|e| Error::Tar(format!("file_type: {e}")))?;
let name_os = entry.file_name();
let name_str = name_os.to_string_lossy();
if matches!(name_str.as_ref(), ".git" | "node_modules") {
continue;
}
let path = entry.path();
if file_type.is_dir() {
self.import_directory_recursive(base, &path, index)?;
continue;
}
if !file_type.is_file() {
continue;
}
let content = std::fs::read(&path)
.map_err(|e| Error::Tar(format!("read {}: {e}", path.display())))?;
#[cfg(unix)]
let executable = {
use std::os::unix::fs::PermissionsExt;
let meta = entry
.metadata()
.map_err(|e| Error::Tar(format!("metadata: {e}")))?;
meta.permissions().mode() & 0o111 != 0
};
#[cfg(not(unix))]
let executable = false;
let rel = path
.strip_prefix(base)
.map_err(|e| Error::Tar(format!("strip_prefix: {e}")))?
.to_string_lossy()
.replace('\\', "/");
let stored = self.import_bytes_gated(&rel, &content, executable)?;
index.insert(rel, stored);
}
Ok(())
}
pub fn import_tarball(&self, tarball_bytes: &[u8]) -> Result<PackageIndex, Error> {
self.import_tarball_reader(tarball_bytes)
}
pub fn import_tarball_reader<R: std::io::Read>(
&self,
compressed_reader: R,
) -> Result<PackageIndex, Error> {
use std::io::Read;
let _diag =
aube_util::diag::Span::new(aube_util::diag::Category::Store, "import_tarball_reader");
let _diag_decode = aube_util::diag::inflight(aube_util::diag::Slot::Decode);
let extract_t0 = std::time::Instant::now();
let gz = flate2::read::GzDecoder::new(compressed_reader);
let capped = CappedReader::new(gz, MAX_TARBALL_DECOMPRESSED_BYTES);
let buffered = std::io::BufReader::with_capacity(256 * 1024, capped);
let mut archive = tar::Archive::new(buffered);
const PIPELINE_CHUNK_SIZE: usize = 64;
let pipelined_disabled = aube_util::env::embedder_env("DISABLE_PIPELINED_IMPORT").is_some();
let parallel_disabled = aube_util::env::embedder_env("DISABLE_PARALLEL_IMPORT").is_some();
let mut staged: Vec<(String, Vec<u8>, bool)> = Vec::new();
let mut entries_seen: usize = 0;
let mut total_uncompressed: u64 = 0;
let mut decode_ns: u128 = 0;
let mut cas_ns: u128 = 0;
let mut index = PackageIndex::default();
let mut staged_count: usize = 0;
let flush_chunk = |chunk: Vec<(String, Vec<u8>, bool)>,
index: &mut PackageIndex,
cas_ns: &mut u128|
-> Result<(), Error> {
if chunk.is_empty() {
return Ok(());
}
let chunk_t0 = std::time::Instant::now();
if parallel_disabled || chunk.len() < PARALLEL_IMPORT_THRESHOLD {
for (rel_path, content, executable) in chunk {
let stored = self.import_bytes_gated(&rel_path, &content, executable)?;
index.insert(rel_path, stored);
}
} else {
use rayon::iter::{
IndexedParallelIterator, IntoParallelIterator, ParallelIterator,
};
const RAYON_TASK_MIN_LEN: usize = 8;
let results: Vec<Result<(String, StoredFile), Error>> = chunk
.into_par_iter()
.with_min_len(RAYON_TASK_MIN_LEN)
.map(|(rel_path, content, executable)| {
self.import_bytes_gated(&rel_path, &content, executable)
.map(|stored| (rel_path, stored))
})
.collect();
for r in results {
let (rel_path, stored) = r?;
index.insert(rel_path, stored);
}
}
*cas_ns += chunk_t0.elapsed().as_nanos();
Ok(())
};
for entry in archive.entries().map_err(|e| Error::Tar(e.to_string()))? {
entries_seen += 1;
if entries_seen > MAX_TARBALL_ENTRIES {
return Err(Error::Tar(format!(
"tarball exceeds entry cap of {MAX_TARBALL_ENTRIES}"
)));
}
let mut entry = entry.map_err(|e| Error::Tar(e.to_string()))?;
let entry_type = entry.header().entry_type();
if entry_type.is_dir()
|| matches!(
entry_type,
tar::EntryType::XGlobalHeader
| tar::EntryType::XHeader
| tar::EntryType::GNULongName
| tar::EntryType::GNULongLink
)
{
continue;
}
if !matches!(
entry_type,
tar::EntryType::Regular | tar::EntryType::Continuous
) {
return Err(Error::Tar(format!(
"tarball entry type {entry_type:?} is not allowed"
)));
}
let declared = entry
.header()
.size()
.map_err(|e| Error::Tar(e.to_string()))?;
if declared > MAX_TARBALL_ENTRY_BYTES {
return Err(Error::Tar(format!(
"tarball entry exceeds per-entry cap: {declared} bytes > {MAX_TARBALL_ENTRY_BYTES}"
)));
}
let raw_path = entry
.path()
.map_err(|e| Error::Tar(e.to_string()))?
.to_path_buf();
let Some(rel_path) = normalize_tar_entry_path(&raw_path)? else {
continue;
};
let mut content = Vec::with_capacity((declared as usize).min(VEC_PREALLOC_CEILING));
let read_t0 = std::time::Instant::now();
(&mut entry)
.take(MAX_TARBALL_ENTRY_BYTES)
.read_to_end(&mut content)
.map_err(|e| Error::Tar(e.to_string()))?;
decode_ns += read_t0.elapsed().as_nanos();
if declared == 0 && !content.is_empty() {
return Err(Error::Tar(format!(
"tarball entry declared 0 bytes but yielded {} bytes",
content.len()
)));
}
let mode = entry.header().mode().unwrap_or(0o644);
let executable = mode & 0o111 != 0;
total_uncompressed = total_uncompressed.saturating_add(content.len() as u64);
staged.push((rel_path, content, executable));
staged_count += 1;
if !pipelined_disabled && staged.len() >= PIPELINE_CHUNK_SIZE {
let chunk = std::mem::take(&mut staged);
flush_chunk(chunk, &mut index, &mut cas_ns)?;
}
}
aube_util::diag::event_lazy(
aube_util::diag::Category::Store,
"tar_extract_complete",
extract_t0.elapsed(),
|| format!(r#"{{"entries":{staged_count},"bytes_uncompressed":{total_uncompressed}}}"#),
);
if aube_util::diag::enabled() {
aube_util::diag::event_lazy(
aube_util::diag::Category::Store,
"gzip_decompress",
std::time::Duration::from_nanos(decode_ns as u64),
|| format!(r#"{{"bytes_uncompressed":{total_uncompressed}}}"#),
);
}
if !staged.is_empty() {
let chunk = std::mem::take(&mut staged);
flush_chunk(chunk, &mut index, &mut cas_ns)?;
}
aube_util::diag::event_lazy(
aube_util::diag::Category::Store,
"cas_import_complete",
std::time::Duration::from_nanos(u64::try_from(cas_ns).unwrap_or(u64::MAX)),
|| {
let pipelined = !pipelined_disabled;
let parallel = !parallel_disabled && staged_count >= PARALLEL_IMPORT_THRESHOLD;
format!(
r#"{{"files":{staged_count},"parallel":{parallel},"pipelined":{pipelined}}}"#
)
},
);
Ok(index)
}
}
const PARALLEL_IMPORT_THRESHOLD: usize = 16;
pub(crate) fn normalize_tar_entry_path(raw: &Path) -> Result<Option<String>, Error> {
use std::path::Component;
let mut components = raw.components().peekable();
while matches!(components.peek(), Some(Component::CurDir)) {
components.next();
}
match components.peek() {
Some(Component::RootDir) => {
return Err(Error::Tar(format!(
"tarball entry path is absolute: {raw:?}"
)));
}
Some(Component::Prefix(_)) => {
return Err(Error::Tar(format!(
"tarball entry path has a Windows drive prefix: {raw:?}"
)));
}
Some(Component::ParentDir) => {
return Err(Error::Tar(format!(
"tarball entry path escapes package root via `..`: {raw:?}"
)));
}
_ => {}
}
components.next();
let mut out = String::with_capacity(raw.as_os_str().len());
for comp in components {
match comp {
Component::Normal(os) => {
let s = os.to_str().ok_or_else(|| {
Error::Tar(format!(
"tarball entry path contains non-UTF-8 bytes: {raw:?}"
))
})?;
if s.is_empty() || s.contains('\0') || s.contains('\\') || s.contains('/') {
return Err(Error::Tar(format!(
"tarball entry path contains a malformed component: {raw:?}"
)));
}
#[cfg(windows)]
{
if s.contains(':') {
return Err(Error::Tar(format!(
"tarball entry path contains a malformed component: {raw:?}"
)));
}
if is_windows_reserved_name(s) {
return Err(Error::Tar(format!(
"tarball entry path contains a Windows reserved device name: {raw:?}"
)));
}
if s.ends_with('.') || s.ends_with(' ') {
return Err(Error::Tar(format!(
"tarball entry path has a trailing dot or space which Windows strips: {raw:?}"
)));
}
if s.bytes().any(|b| b < 0x20) {
return Err(Error::Tar(format!(
"tarball entry path contains control characters: {raw:?}"
)));
}
}
if !out.is_empty() {
out.push('/');
}
out.push_str(s);
}
Component::ParentDir => {
return Err(Error::Tar(format!(
"tarball entry path escapes package root via `..`: {raw:?}"
)));
}
Component::RootDir => {
return Err(Error::Tar(format!(
"tarball entry path is absolute: {raw:?}"
)));
}
Component::Prefix(_) => {
return Err(Error::Tar(format!(
"tarball entry path has a Windows drive prefix: {raw:?}"
)));
}
Component::CurDir => {}
}
}
if out.is_empty() {
Ok(None)
} else {
Ok(Some(out))
}
}
#[cfg(windows)]
fn is_windows_reserved_name(name: &str) -> bool {
let stem = name.split_once('.').map(|(a, _)| a).unwrap_or(name);
let upper = stem.to_ascii_uppercase();
matches!(
upper.as_str(),
"CON"
| "PRN"
| "AUX"
| "NUL"
| "COM1"
| "COM2"
| "COM3"
| "COM4"
| "COM5"
| "COM6"
| "COM7"
| "COM8"
| "COM9"
| "LPT1"
| "LPT2"
| "LPT3"
| "LPT4"
| "LPT5"
| "LPT6"
| "LPT7"
| "LPT8"
| "LPT9"
)
}
const VEC_PREALLOC_CEILING: usize = 64 * 1024;
pub(crate) struct CappedReader<R: std::io::Read> {
inner: R,
remaining: u64,
}
impl<R: std::io::Read> CappedReader<R> {
pub(crate) fn new(inner: R, cap: u64) -> Self {
Self {
inner,
remaining: cap,
}
}
}
impl<R: std::io::Read> std::io::Read for CappedReader<R> {
fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
if buf.is_empty() {
return Ok(0);
}
if self.remaining == 0 {
return Err(std::io::Error::new(
std::io::ErrorKind::InvalidData,
format!(
"tarball decompression exceeds archive cap of {MAX_TARBALL_DECOMPRESSED_BYTES} bytes"
),
));
}
let want = buf.len().min(self.remaining as usize);
let n = self.inner.read(&mut buf[..want])?;
self.remaining -= n as u64;
Ok(n)
}
}
#[cfg(not(test))]
pub(crate) const MAX_TARBALL_DECOMPRESSED_BYTES: u64 = 1 << 30;
#[cfg(test)]
pub(crate) const MAX_TARBALL_DECOMPRESSED_BYTES: u64 = 1 << 20;
#[cfg(not(test))]
pub(crate) const MAX_TARBALL_ENTRY_BYTES: u64 = 512 << 20;
#[cfg(test)]
pub(crate) const MAX_TARBALL_ENTRY_BYTES: u64 = 1 << 20;
#[cfg(not(test))]
pub(crate) const MAX_TARBALL_ENTRIES: usize = 200_000;
#[cfg(test)]
pub(crate) const MAX_TARBALL_ENTRIES: usize = 64;
#[cfg(test)]
mod directory_fingerprint_tests {
use super::*;
#[test]
fn directory_fingerprint_matches_imported_index() {
let temp = tempfile::tempdir().unwrap();
let source = temp.path().join("source");
std::fs::create_dir_all(source.join("lib")).unwrap();
std::fs::create_dir_all(source.join("node_modules/ignored")).unwrap();
std::fs::write(source.join("package.json"), br#"{"name":"local"}"#).unwrap();
std::fs::write(source.join("lib/index.js"), b"module.exports = 'v1';\n").unwrap();
std::fs::write(source.join("node_modules/ignored/index.js"), b"ignored\n").unwrap();
let store = Store::at(temp.path().join("store"));
let index = store.import_directory(&source).unwrap();
let (content_hash, metadata_hash) = directory_fingerprints(&source).unwrap();
assert_eq!(content_hash, crate::index_content_fingerprint(&index));
assert_eq!(
metadata_hash,
directory_metadata_fingerprint(&source).unwrap()
);
let before_content = content_hash;
std::fs::write(source.join("lib/index.js"), b"module.exports = 'v2';\n").unwrap();
let after_content = directory_content_fingerprint(&source).unwrap();
assert_ne!(before_content, after_content);
std::fs::write(source.join("lib/added.js"), b"added\n").unwrap();
assert_ne!(
metadata_hash,
directory_metadata_fingerprint(&source).unwrap()
);
}
}