diff --git a/Cargo.lock b/Cargo.lock index 35c5e9c..9a41996 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -665,6 +665,17 @@ dependencies = [ "tracing-subscriber", ] +[[package]] +name = "crank-artifacts" +version = "0.3.1" +dependencies = [ + "libc", + "rand 0.10.2", + "sha2 0.10.9", + "thiserror 2.0.18", + "url", +] + [[package]] name = "crank-community-auth" version = "0.3.1" diff --git a/Cargo.toml b/Cargo.toml index 8dc88e4..66477f5 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -16,6 +16,7 @@ members = [ "crates/crank-test-support", "crates/crank-trace", "crates/crank-adapter-rest", + "crates/crank-artifacts", ] resolver = "3" diff --git a/crates/crank-artifacts/Cargo.toml b/crates/crank-artifacts/Cargo.toml new file mode 100644 index 0000000..1352718 --- /dev/null +++ b/crates/crank-artifacts/Cargo.toml @@ -0,0 +1,14 @@ +[package] +name = "crank-artifacts" +edition.workspace = true +license.workspace = true +rust-version.workspace = true +publish.workspace = true +version.workspace = true + +[dependencies] +libc = "0.2" +rand.workspace = true +sha2.workspace = true +thiserror.workspace = true +url.workspace = true diff --git a/crates/crank-artifacts/src/error.rs b/crates/crank-artifacts/src/error.rs new file mode 100644 index 0000000..9d00c7a --- /dev/null +++ b/crates/crank-artifacts/src/error.rs @@ -0,0 +1,23 @@ +use thiserror::Error; + +/// Stable, redacted failures from the artifact boundary. +/// +/// This intentionally carries neither an operating-system source nor a path: +/// those details belong only in the composing process's private logs. +#[derive(Clone, Copy, Debug, Eq, Error, PartialEq)] +pub enum ArtifactError { + #[error("artifact reference is invalid")] + InvalidReference, + #[error("artifact source is empty")] + EmptySource, + #[error("artifact source exceeds the configured maximum")] + SourceTooLarge, + #[error("artifact was not found")] + NotFound, + #[error("artifact integrity verification failed")] + Integrity, + #[error("artifact storage root or entry is unsafe")] + UnsafeRoot, + #[error("artifact storage is unavailable")] + Storage, +} diff --git a/crates/crank-artifacts/src/housekeeping.rs b/crates/crank-artifacts/src/housekeeping.rs new file mode 100644 index 0000000..ac628dc --- /dev/null +++ b/crates/crank-artifacts/src/housekeeping.rs @@ -0,0 +1,234 @@ +use std::{ + os::fd::AsRawFd, + time::{Duration, SystemTime, UNIX_EPOCH}, +}; + +use crate::{ + ArtifactError, ArtifactStore, + store::{RootLock, ensure_root_unchanged, fsync_fd, open_existing_dir, unlinkat}, +}; + +/// Opaque, single-use evidence that a particular stale temporary inode was +/// observed under this store's pinned root. It is not a blob reference. +#[derive(Debug)] +pub struct StaleTemp { + name: String, + shard: String, + root_dev: u64, + root_ino: u64, + dev: u64, + ino: u64, + modified_seconds: i64, + modified_nanoseconds: i64, + grace: Duration, +} + +#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)] +pub struct TempScan { + pub scanned: usize, + pub omitted: usize, +} + +impl ArtifactStore { + /// Streams temporary entries, reading at most `scan_budget` directory + /// entries and returning at most `result_limit` stale capabilities. + pub fn scan_stale_temps( + &self, + grace: Duration, + scan_budget: usize, + result_limit: usize, + ) -> Result<(TempScan, Vec), ArtifactError> { + let root = self.root()?; + let _lock = RootLock::shared(root)?; + ensure_root_unchanged(root)?; + let sha = match open_existing_dir(root.fd.as_raw_fd(), b"sha256") { + Ok(fd) => fd, + Err(ArtifactError::NotFound) => return Ok((TempScan::default(), Vec::new())), + Err(error) => return Err(error), + }; + let mut report = TempScan::default(); + let mut remaining = scan_budget; + let mut result = Vec::new(); + for shard in (0_u8..=255).map(|value| format!("{value:02x}")) { + if remaining == 0 { + break; + } + let shard_fd = match open_existing_dir(sha.as_raw_fd(), shard.as_bytes()) { + Ok(fd) => fd, + Err(ArtifactError::NotFound) => continue, + Err(error) => return Err(error), + }; + for name in list_names(shard_fd.as_raw_fd(), &mut remaining, &mut report)? { + if !valid_temp_name(&name) { + continue; + } + #[cfg(debug_assertions)] + let _ = crate::test_support::checkpoint("housekeeping_before_stat"); + let stat = match nofollow_stat(shard_fd.as_raw_fd(), &name) { + Ok(stat) => stat, + Err(ArtifactError::NotFound) => continue, + Err(error) => return Err(error), + }; + if (stat.st_mode & libc::S_IFMT) != libc::S_IFREG || stat.st_nlink != 1 { + continue; + } + let modified = UNIX_EPOCH + .checked_add(Duration::new( + stat.st_mtime.max(0) as u64, + stat.st_mtime_nsec.max(0) as u32, + )) + .unwrap_or(SystemTime::UNIX_EPOCH); + if SystemTime::now() + .duration_since(modified) + .is_ok_and(|age| age >= grace) + { + if result.len() == result_limit { + report.omitted += 1; + continue; + } + result.push(StaleTemp { + name, + shard: shard.clone(), + root_dev: root.dev, + root_ino: root.ino, + dev: stat.st_dev, + ino: stat.st_ino, + modified_seconds: stat.st_mtime, + modified_nanoseconds: stat.st_mtime_nsec, + grace, + }); + } + } + } + Ok((report, result)) + } + + /// Deletes only a previously discovered temporary inode after an exclusive + /// cooperative root lock and inode revalidation, then fsyncs its shard. + pub fn delete_stale_temp(&self, candidate: StaleTemp) -> Result<(), ArtifactError> { + let root = self.root()?; + let _lock = RootLock::exclusive(root)?; + ensure_root_unchanged(root)?; + if candidate.root_dev != root.dev || candidate.root_ino != root.ino { + return Err(ArtifactError::UnsafeRoot); + } + if !valid_temp_name(&candidate.name) + || candidate.shard.len() != 2 + || !candidate.shard.bytes().all(is_lower_hex) + { + return Err(ArtifactError::UnsafeRoot); + } + let sha = open_existing_dir(root.fd.as_raw_fd(), b"sha256")?; + let shard = open_existing_dir(sha.as_raw_fd(), candidate.shard.as_bytes())?; + let stat = nofollow_stat(shard.as_raw_fd(), &candidate.name)?; + if stat.st_dev != candidate.dev + || stat.st_ino != candidate.ino + || stat.st_mtime != candidate.modified_seconds + || stat.st_mtime_nsec != candidate.modified_nanoseconds + || (stat.st_mode & libc::S_IFMT) != libc::S_IFREG + || stat.st_nlink != 1 + { + return Err(ArtifactError::UnsafeRoot); + } + let modified = UNIX_EPOCH + .checked_add(Duration::new( + stat.st_mtime.max(0) as u64, + stat.st_mtime_nsec.max(0) as u32, + )) + .unwrap_or(SystemTime::UNIX_EPOCH); + if !SystemTime::now() + .duration_since(modified) + .is_ok_and(|age| age >= candidate.grace) + { + return Err(ArtifactError::UnsafeRoot); + } + unlinkat(shard.as_raw_fd(), candidate.name.as_bytes())?; + fsync_fd(shard.as_raw_fd()) + } +} + +fn list_names( + parent: i32, + remaining: &mut usize, + report: &mut TempScan, +) -> Result, ArtifactError> { + let dot = std::ffi::CString::new(".").map_err(|_| ArtifactError::Storage)?; + let raw = unsafe { + libc::openat( + parent, + dot.as_ptr(), + libc::O_RDONLY | libc::O_DIRECTORY | libc::O_CLOEXEC, + ) + }; + if raw < 0 { + return Err(ArtifactError::Storage); + } + let directory = unsafe { libc::fdopendir(raw) }; + if directory.is_null() { + unsafe { + libc::close(raw); + } + return Err(ArtifactError::Storage); + } + let mut names = Vec::new(); + loop { + if *remaining == 0 { + break; + } + unsafe { + *libc::__errno_location() = 0; + } + let entry = unsafe { libc::readdir(directory) }; + if entry.is_null() { + if unsafe { *libc::__errno_location() } != 0 { + unsafe { + libc::closedir(directory); + } + return Err(ArtifactError::Storage); + } + break; + } + let name = unsafe { std::ffi::CStr::from_ptr((*entry).d_name.as_ptr()) }.to_bytes(); + if name == b"." || name == b".." { + continue; + } + report.scanned += 1; + *remaining -= 1; + if let Ok(name) = std::str::from_utf8(name) { + names.push(name.to_owned()); + } + } + unsafe { + libc::closedir(directory); + } + Ok(names) +} +fn valid_temp_name(name: &str) -> bool { + let Some(rest) = name.strip_prefix(ArtifactStore::temp_prefix()) else { + return false; + }; + let mut fields = rest.split('-'); + let (Some(nonce), Some(pid), Some(sequence), None) = + (fields.next(), fields.next(), fields.next(), fields.next()) + else { + return false; + }; + nonce.len() == 32 + && nonce.bytes().all(is_lower_hex) + && !pid.is_empty() + && pid.bytes().all(|byte| byte.is_ascii_digit()) + && !sequence.is_empty() + && sequence.bytes().all(|byte| byte.is_ascii_digit()) +} +fn is_lower_hex(byte: u8) -> bool { + byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte) +} +fn nofollow_stat(parent: i32, name: &str) -> Result { + let name = crate::store::c_name(name.as_bytes())?; + let mut stat = unsafe { std::mem::zeroed() }; + if unsafe { libc::fstatat(parent, name.as_ptr(), &mut stat, libc::AT_SYMLINK_NOFOLLOW) } == 0 { + Ok(stat) + } else { + Err(crate::store::classify_errno()) + } +} diff --git a/crates/crank-artifacts/src/legacy.rs b/crates/crank-artifacts/src/legacy.rs new file mode 100644 index 0000000..44fd66f --- /dev/null +++ b/crates/crank-artifacts/src/legacy.rs @@ -0,0 +1,2 @@ +//! The legacy reader lives on [`crate::ArtifactStore`] so it shares the same +//! pinned-FD and digest-verification primitives as ordinary reads. diff --git a/crates/crank-artifacts/src/lib.rs b/crates/crank-artifacts/src/lib.rs new file mode 100644 index 0000000..d6173a0 --- /dev/null +++ b/crates/crank-artifacts/src/lib.rs @@ -0,0 +1,23 @@ +//! A small, Linux-only immutable content-addressed artifact store. +//! +//! The directory file descriptor opened by [`ArtifactStore::open`] is the +//! authority for all operations. Paths supplied by callers are never used for +//! blob I/O, and the public API deliberately exposes only opaque digests. + +#[cfg(not(target_os = "linux"))] +compile_error!("crank-artifacts requires Linux"); + +mod error; +mod housekeeping; +mod legacy; +mod model; +mod store; + +#[cfg(debug_assertions)] +#[doc(hidden)] +pub mod test_support; + +pub use error::ArtifactError; +pub use housekeeping::{StaleTemp, TempScan}; +pub use model::{ArtifactRef, MAX_ARTIFACT_BYTES, MAX_SOURCE_BYTES, StoredArtifact}; +pub use store::ArtifactStore; diff --git a/crates/crank-artifacts/src/model.rs b/crates/crank-artifacts/src/model.rs new file mode 100644 index 0000000..a711485 --- /dev/null +++ b/crates/crank-artifacts/src/model.rs @@ -0,0 +1,68 @@ +use std::{fmt, str::FromStr}; + +use crate::ArtifactError; + +/// The largest source accepted by the OpenAPI ingress contract. +pub const MAX_ARTIFACT_BYTES: usize = 256 * 1024; +/// Backwards-compatible name for the source-upload limit. +pub const MAX_SOURCE_BYTES: usize = MAX_ARTIFACT_BYTES; +const PREFIX: &str = "sha256:"; + +/// Canonical, opaque SHA-256 reference (`sha256:` plus 64 lowercase hex +/// characters). It cannot represent a filesystem name or path. +#[derive(Clone, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)] +pub struct ArtifactRef(String); + +impl ArtifactRef { + pub fn parse(value: &str) -> Result { + let digest = value + .strip_prefix(PREFIX) + .ok_or(ArtifactError::InvalidReference)?; + Self::from_digest_hex(digest) + } + + pub fn from_digest_hex(digest: &str) -> Result { + if digest.len() != 64 + || !digest + .as_bytes() + .iter() + .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(byte)) + { + return Err(ArtifactError::InvalidReference); + } + Ok(Self(format!("{PREFIX}{digest}"))) + } + + pub fn as_str(&self) -> &str { + &self.0 + } + + pub fn digest_hex(&self) -> &str { + &self.0[PREFIX.len()..] + } + + pub(crate) fn shard(&self) -> &str { + // Construction proves the digest has 64 ASCII bytes. + &self.digest_hex()[..2] + } +} + +impl fmt::Display for ArtifactRef { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter.write_str(self.as_str()) + } +} + +impl FromStr for ArtifactRef { + type Err = ArtifactError; + + fn from_str(value: &str) -> Result { + Self::parse(value) + } +} + +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct StoredArtifact { + pub artifact_ref: ArtifactRef, + pub size_bytes: usize, +} diff --git a/crates/crank-artifacts/src/store.rs b/crates/crank-artifacts/src/store.rs new file mode 100644 index 0000000..9bf5162 --- /dev/null +++ b/crates/crank-artifacts/src/store.rs @@ -0,0 +1,663 @@ +use std::{ + ffi::CString, + fs::File, + io::{self, Read, Write}, + os::{ + fd::{AsRawFd, FromRawFd, OwnedFd}, + unix::ffi::OsStrExt, + }, + path::{Component, Path}, + sync::atomic::{AtomicU64, Ordering}, +}; + +use rand::random; +use sha2::{Digest, Sha256}; + +use crate::{ArtifactError, ArtifactRef, MAX_ARTIFACT_BYTES, StoredArtifact}; + +const SHA_DIR: &[u8] = b"sha256"; +const TEMP_PREFIX: &str = ".crank-artifact-tmp-v1-"; +const RESOLVE_NO_SYMLINKS: u64 = 0x04; +static TEMP_SEQUENCE: AtomicU64 = AtomicU64::new(0); + +#[derive(Debug)] +pub(crate) struct Root { + pub(crate) fd: OwnedFd, + pub(crate) dev: u64, + pub(crate) ino: u64, +} + +#[derive(Debug)] +struct LegacyRoot { + root: Root, + configured_path: std::path::PathBuf, +} + +/// Immutable content-addressed store anchored to one trusted directory inode. +#[derive(Debug)] +pub struct ArtifactStore { + root: Result, + legacy: Option>, + nonce: u128, +} + +impl Clone for ArtifactStore { + fn clone(&self) -> Self { + // A fresh open file description makes flock coordination independent + // between clones while retaining the original directory authority. + let root = self + .root + .as_ref() + .map_err(|error| *error) + .and_then(|root| duplicate_root(root).map_err(|_| ArtifactError::Storage)); + let legacy = self.legacy.as_ref().map(|legacy| { + legacy.as_ref().map_err(|error| *error).and_then(|legacy| { + duplicate_root(&legacy.root) + .map(|root| LegacyRoot { + root, + configured_path: legacy.configured_path.clone(), + }) + .map_err(|_| ArtifactError::Storage) + }) + }); + Self { + root, + legacy, + nonce: self.nonce, + } + } +} + +impl ArtifactStore { + /// Opens an already provisioned, private artifact root. The root is never + /// reopened by pathname after this point. + pub fn open(root: impl AsRef) -> Result { + Ok(Self { + root: Ok(open_trusted_root(root.as_ref())?), + legacy: None, + nonce: random(), + }) + } + + /// Compatibility constructor. I/O returns the opening error fail-closed. + pub fn new(root: impl AsRef) -> Self { + Self { + root: open_trusted_root(root.as_ref()), + legacy: None, + nonce: random(), + } + } + + /// Enables the explicit, read-only compatibility reader for one pinned + /// legacy root. It never migrates, rewrites, or deletes legacy files. + pub fn with_legacy_root(mut self, root: impl AsRef) -> Self { + let configured_path = normalize_absolute_path(root.as_ref()); + self.legacy = Some(configured_path.and_then(|configured_path| { + open_trusted_root(root.as_ref()).map(|root_fd| LegacyRoot { + root: root_fd, + configured_path, + }) + })); + self + } + + /// Stores a complete bounded byte slice durably and returns its SHA-256 ref. + pub fn put(&self, bytes: &[u8]) -> Result { + self.put_reader(&mut io::Cursor::new(bytes)) + } + + /// Historical spelling retained for source compatibility; bytes are not + /// restricted to UTF-8 at this filesystem boundary. + pub fn put_utf8(&self, bytes: &[u8]) -> Result { + self.put(bytes) + } + + /// Streams at most [`MAX_ARTIFACT_BYTES`] from `reader`, detecting an + /// oversized source before publication even with short or interrupted reads. + pub fn put_reader(&self, reader: &mut impl Read) -> Result { + let mut bytes = Vec::with_capacity(8 * 1024); + let mut hasher = Sha256::new(); + let mut buffer = [0_u8; 8192]; + loop { + match reader.read(&mut buffer) { + Ok(0) => break, + Ok(count) => { + if bytes + .len() + .checked_add(count) + .ok_or(ArtifactError::SourceTooLarge)? + > MAX_ARTIFACT_BYTES + { + return Err(ArtifactError::SourceTooLarge); + } + hasher.update(&buffer[..count]); + bytes.extend_from_slice(&buffer[..count]); + } + Err(error) if error.kind() == io::ErrorKind::Interrupted => continue, + Err(_) => return Err(ArtifactError::Storage), + } + } + if bytes.is_empty() { + return Err(ArtifactError::EmptySource); + } + let artifact_ref = ArtifactRef::from_digest_hex(&format!("{:x}", hasher.finalize()))?; + let root = self.root()?; + let _lock = RootLock::shared(root)?; + ensure_root_unchanged(root)?; + let sha = open_or_create_dir(root.fd.as_raw_fd(), SHA_DIR)?; + let shard = open_or_create_dir(sha.as_raw_fd(), artifact_ref.shard().as_bytes())?; + self.publish(root, shard.as_raw_fd(), &artifact_ref, &bytes)?; + Ok(StoredArtifact { + artifact_ref, + size_bytes: bytes.len(), + }) + } + + /// Reads a blob only after validating its inode properties, bounded length, + /// and complete SHA-256 digest. + pub fn read(&self, artifact_ref: &ArtifactRef) -> Result, ArtifactError> { + let root = self.root()?; + let _lock = RootLock::shared(root)?; + ensure_root_unchanged(root)?; + let sha = open_existing_dir(root.fd.as_raw_fd(), SHA_DIR)?; + let shard = open_existing_dir(sha.as_raw_fd(), artifact_ref.shard().as_bytes())?; + let fd = open_existing_file(shard.as_raw_fd(), artifact_ref.digest_hex().as_bytes())?; + read_verified(&fd, artifact_ref) + } + + /// Digest-verified `file://` compatibility read below the explicitly + /// configured legacy root. A URL cannot select another filesystem root. + pub fn read_legacy_file_url( + &self, + value: &str, + expected: &ArtifactRef, + ) -> Result, ArtifactError> { + let legacy = self + .legacy + .as_ref() + .ok_or(ArtifactError::UnsafeRoot)? + .as_ref() + .map_err(|error| *error)?; + let _lock = RootLock::shared(&legacy.root)?; + ensure_root_unchanged(&legacy.root)?; + let url = url::Url::parse(value).map_err(|_| ArtifactError::InvalidReference)?; + if url.scheme() != "file" || url.host_str().is_some() { + return Err(ArtifactError::InvalidReference); + } + let candidate = url + .to_file_path() + .map_err(|_| ArtifactError::InvalidReference)?; + let relative = candidate + .strip_prefix(&legacy.configured_path) + .map_err(|_| ArtifactError::UnsafeRoot)?; + let mut fd = duplicate_fd(legacy.root.fd.as_raw_fd())?; + let mut components = relative.components().peekable(); + while let Some(component) = components.next() { + let Component::Normal(name) = component else { + return Err(ArtifactError::UnsafeRoot); + }; + fd = if components.peek().is_some() { + open_legacy_dir(fd.as_raw_fd(), name.as_bytes())? + } else { + open_legacy_file(fd.as_raw_fd(), name.as_bytes())? + }; + } + read_legacy_verified(&fd, expected) + } + + fn publish( + &self, + root: &Root, + shard_fd: i32, + artifact_ref: &ArtifactRef, + bytes: &[u8], + ) -> Result<(), ArtifactError> { + let final_name = artifact_ref.digest_hex().as_bytes(); + match open_existing_file(shard_fd, final_name) { + Ok(file) => { + let existing = read_verified(&file, artifact_ref)?; + if existing == bytes { + checkpointed("dedupe_directory_fsync", || fsync_fd(shard_fd))?; + return Ok(()); + } + return Err(ArtifactError::Integrity); + } + Err(ArtifactError::NotFound) => {} + Err(error) => return Err(error), + } + + let temp_name = self.temp_name(); + let temp = create_temp(shard_fd, temp_name.as_bytes())?; + let result = (|| { + checkpointed("write", || write_all(&temp, bytes))?; + checkpointed("file_fsync", || fsync_fd(temp.as_raw_fd()))?; + chmod_read_only(temp.as_raw_fd())?; + fsync_fd(temp.as_raw_fd())?; + ensure_root_unchanged(root)?; + match checkpointed("publish", || { + rename_no_replace(shard_fd, temp_name.as_bytes(), final_name) + })? { + true => checkpointed("directory_fsync", || fsync_fd(shard_fd)), + false => { + let file = open_existing_file(shard_fd, final_name)?; + let existing = read_verified(&file, artifact_ref)?; + if existing == bytes { + checkpointed("dedupe_directory_fsync", || fsync_fd(shard_fd)) + } else { + Err(ArtifactError::Integrity) + } + } + } + })(); + let _ = unlinkat(shard_fd, temp_name.as_bytes()); + result + } + + pub(crate) fn root(&self) -> Result<&Root, ArtifactError> { + self.root.as_ref().map_err(|error| *error) + } + pub(crate) fn temp_prefix() -> &'static str { + TEMP_PREFIX + } + pub(crate) fn temp_name(&self) -> String { + format!( + "{TEMP_PREFIX}{:032x}-{}-{}", + self.nonce, + std::process::id(), + TEMP_SEQUENCE.fetch_add(1, Ordering::Relaxed) + ) + } +} + +fn normalize_absolute_path(path: &Path) -> Result { + if !path.is_absolute() { + return Err(ArtifactError::UnsafeRoot); + } + let mut normalized = std::path::PathBuf::from("/"); + for component in path.components() { + match component { + Component::RootDir | Component::CurDir => {} + Component::Normal(part) => normalized.push(part), + Component::ParentDir => { + if !normalized.pop() { + return Err(ArtifactError::UnsafeRoot); + } + } + Component::Prefix(_) => return Err(ArtifactError::UnsafeRoot), + } + } + Ok(normalized) +} + +pub(crate) fn c_name(value: &[u8]) -> Result { + CString::new(value).map_err(|_| ArtifactError::InvalidReference) +} +pub(crate) fn duplicate_fd(fd: i32) -> Result { + let duplicate = unsafe { libc::fcntl(fd, libc::F_DUPFD_CLOEXEC, 0) }; + if duplicate < 0 { + Err(ArtifactError::Storage) + } else { + Ok(unsafe { OwnedFd::from_raw_fd(duplicate) }) + } +} +fn duplicate_root(root: &Root) -> Result { + let dot = c_name(b".")?; + let raw = unsafe { + libc::openat( + root.fd.as_raw_fd(), + dot.as_ptr(), + libc::O_RDONLY | libc::O_DIRECTORY | libc::O_CLOEXEC | libc::O_NOFOLLOW, + ) + }; + if raw < 0 { + return Err(classify_errno()); + } + let fd = unsafe { OwnedFd::from_raw_fd(raw) }; + Ok(Root { + fd, + dev: root.dev, + ino: root.ino, + }) +} + +fn open_trusted_root(path: &Path) -> Result { + let path = CString::new(path.as_os_str().as_bytes()).map_err(|_| ArtifactError::UnsafeRoot)?; + let how = OpenHow { + flags: (libc::O_RDONLY | libc::O_DIRECTORY | libc::O_CLOEXEC | libc::O_NOFOLLOW) as u64, + mode: 0, + resolve: RESOLVE_NO_SYMLINKS, + }; + let fd = unsafe { + libc::syscall( + libc::SYS_openat2, + libc::AT_FDCWD, + path.as_ptr(), + &how, + std::mem::size_of::(), + ) as i32 + }; + if fd < 0 { + return Err(classify_errno()); + } + let fd = unsafe { OwnedFd::from_raw_fd(fd) }; + let stat = stat_fd(fd.as_raw_fd())?; + check_private_dir(&stat)?; + Ok(Root { + fd, + dev: stat.st_dev, + ino: stat.st_ino, + }) +} + +#[repr(C)] +struct OpenHow { + flags: u64, + mode: u64, + resolve: u64, +} + +pub(crate) fn ensure_root_unchanged(root: &Root) -> Result<(), ArtifactError> { + let stat = stat_fd(root.fd.as_raw_fd())?; + if stat.st_dev != root.dev || stat.st_ino != root.ino { + return Err(ArtifactError::UnsafeRoot); + } + check_private_dir(&stat) +} +fn check_private_dir(stat: &libc::stat) -> Result<(), ArtifactError> { + if stat.st_uid != unsafe { libc::geteuid() } + || (stat.st_mode & libc::S_IFMT) != libc::S_IFDIR + || (stat.st_mode & 0o777) != 0o700 + { + return Err(ArtifactError::UnsafeRoot); + } + Ok(()) +} +fn stat_fd(fd: i32) -> Result { + let mut stat = unsafe { std::mem::zeroed() }; + if unsafe { libc::fstat(fd, &mut stat) } != 0 { + Err(ArtifactError::Storage) + } else { + Ok(stat) + } +} +fn check_file(stat: &libc::stat) -> Result<(), ArtifactError> { + if (stat.st_mode & libc::S_IFMT) != libc::S_IFREG + || stat.st_uid != unsafe { libc::geteuid() } + || stat.st_nlink != 1 + || (stat.st_mode & 0o222) != 0 + { + return Err(ArtifactError::Integrity); + } + Ok(()) +} +pub(crate) fn open_or_create_dir(parent: i32, name: &[u8]) -> Result { + let fd = match open_existing_dir(parent, name) { + Ok(fd) => fd, + Err(ArtifactError::NotFound) => { + let name = c_name(name)?; + let code = unsafe { libc::mkdirat(parent, name.as_ptr(), 0o700) }; + if code != 0 && unsafe { *libc::__errno_location() } != libc::EEXIST { + return Err(ArtifactError::Storage); + } + open_existing_dir(parent, name.as_bytes())? + } + Err(error) => return Err(error), + }; + checkpointed("mkdir_parent_fsync", || fsync_fd(parent))?; + Ok(fd) +} +pub(crate) fn open_existing_dir(parent: i32, name: &[u8]) -> Result { + let name = c_name(name)?; + let fd = unsafe { + libc::openat( + parent, + name.as_ptr(), + libc::O_RDONLY | libc::O_DIRECTORY | libc::O_CLOEXEC | libc::O_NOFOLLOW, + ) + }; + if fd < 0 { + return Err(classify_errno()); + } + let fd = unsafe { OwnedFd::from_raw_fd(fd) }; + check_private_dir(&stat_fd(fd.as_raw_fd())?)?; + Ok(fd) +} +pub(crate) fn open_existing_file(parent: i32, name: &[u8]) -> Result { + let name = c_name(name)?; + let fd = unsafe { + libc::openat( + parent, + name.as_ptr(), + libc::O_RDONLY | libc::O_CLOEXEC | libc::O_NOFOLLOW | libc::O_NONBLOCK, + ) + }; + if fd < 0 { + return Err(classify_errno()); + } + let fd = unsafe { OwnedFd::from_raw_fd(fd) }; + check_file(&stat_fd(fd.as_raw_fd())?)?; + Ok(fd) +} +fn open_legacy_dir(parent: i32, name: &[u8]) -> Result { + let name = c_name(name)?; + let raw = unsafe { + libc::openat( + parent, + name.as_ptr(), + libc::O_RDONLY | libc::O_DIRECTORY | libc::O_CLOEXEC | libc::O_NOFOLLOW, + ) + }; + if raw < 0 { + return Err(classify_errno()); + } + let fd = unsafe { OwnedFd::from_raw_fd(raw) }; + if (stat_fd(fd.as_raw_fd())?.st_mode & libc::S_IFMT) != libc::S_IFDIR { + return Err(ArtifactError::UnsafeRoot); + } + Ok(fd) +} +fn open_legacy_file(parent: i32, name: &[u8]) -> Result { + let name = c_name(name)?; + let raw = unsafe { + libc::openat( + parent, + name.as_ptr(), + libc::O_RDONLY | libc::O_CLOEXEC | libc::O_NOFOLLOW | libc::O_NONBLOCK, + ) + }; + if raw < 0 { + return Err(classify_errno()); + } + let fd = unsafe { OwnedFd::from_raw_fd(raw) }; + if (stat_fd(fd.as_raw_fd())?.st_mode & libc::S_IFMT) != libc::S_IFREG { + return Err(ArtifactError::Integrity); + } + Ok(fd) +} +fn create_temp(parent: i32, name: &[u8]) -> Result { + let name = c_name(name)?; + let fd = unsafe { + libc::openat( + parent, + name.as_ptr(), + libc::O_WRONLY | libc::O_CREAT | libc::O_EXCL | libc::O_CLOEXEC | libc::O_NOFOLLOW, + 0o600, + ) + }; + if fd < 0 { + return Err(classify_errno()); + } + Ok(unsafe { OwnedFd::from_raw_fd(fd) }) +} +fn write_all(fd: &OwnedFd, bytes: &[u8]) -> Result<(), ArtifactError> { + let mut file = File::from(duplicate_fd(fd.as_raw_fd())?); + let result = file.write_all(bytes).map_err(|_| ArtifactError::Storage); + drop(file); + result +} + +/// Debug/test-only fault seam coupled to one syscall wrapper. `Fail` models a +/// syscall that did not run; crash/hold actions happen only after success. +fn checkpointed( + stage: &str, + operation: impl FnOnce() -> Result, +) -> Result { + #[cfg(debug_assertions)] + { + if crate::test_support::action(stage) == Some(crate::test_support::FaultAction::Fail) { + return Err(ArtifactError::Storage); + } + } + let result = operation()?; + #[cfg(debug_assertions)] + { + match crate::test_support::checkpoint(stage) { + Some(crate::test_support::FaultAction::Exit) => { + // Do not run Rust destructors: this models abrupt process + // death, rather than an ordinary in-process error path. + unsafe { libc::_exit(86) }; + } + Some(crate::test_support::FaultAction::Fail) => unreachable!("handled before syscall"), + Some(crate::test_support::FaultAction::Hold) => {} + None => {} + } + } + let _ = stage; + Ok(result) +} +pub(crate) fn read_verified( + fd: &OwnedFd, + expected: &ArtifactRef, +) -> Result, ArtifactError> { + check_file(&stat_fd(fd.as_raw_fd())?)?; + let length = + usize::try_from(stat_fd(fd.as_raw_fd())?.st_size).map_err(|_| ArtifactError::Integrity)?; + if length == 0 || length > MAX_ARTIFACT_BYTES { + return Err(ArtifactError::Integrity); + } + let mut file = File::from(duplicate_fd(fd.as_raw_fd())?); + let mut bytes = Vec::with_capacity(length); + let result = Read::take(&mut file, (MAX_ARTIFACT_BYTES + 1) as u64) + .read_to_end(&mut bytes) + .map_err(|_| ArtifactError::Storage); + drop(file); + result?; + if bytes.len() > MAX_ARTIFACT_BYTES + || bytes.len() != length + || format!("{:x}", Sha256::digest(&bytes)) != expected.digest_hex() + { + return Err(ArtifactError::Integrity); + } + Ok(bytes) +} + +fn read_legacy_verified(fd: &OwnedFd, expected: &ArtifactRef) -> Result, ArtifactError> { + let stat = stat_fd(fd.as_raw_fd())?; + if (stat.st_mode & libc::S_IFMT) != libc::S_IFREG { + return Err(ArtifactError::Integrity); + } + read_bounded_digest(fd, expected, stat.st_size) +} + +fn read_bounded_digest( + fd: &OwnedFd, + expected: &ArtifactRef, + raw_length: libc::off_t, +) -> Result, ArtifactError> { + let length = usize::try_from(raw_length).map_err(|_| ArtifactError::Integrity)?; + if length == 0 || length > MAX_ARTIFACT_BYTES { + return Err(ArtifactError::Integrity); + } + let mut file = File::from(duplicate_fd(fd.as_raw_fd())?); + let mut bytes = Vec::with_capacity(length); + let result = Read::take(&mut file, (MAX_ARTIFACT_BYTES + 1) as u64) + .read_to_end(&mut bytes) + .map_err(|_| ArtifactError::Storage); + drop(file); + result?; + if bytes.len() > MAX_ARTIFACT_BYTES + || bytes.len() != length + || format!("{:x}", Sha256::digest(&bytes)) != expected.digest_hex() + { + return Err(ArtifactError::Integrity); + } + Ok(bytes) +} +pub(crate) fn fsync_fd(fd: i32) -> Result<(), ArtifactError> { + if unsafe { libc::fsync(fd) } == 0 { + Ok(()) + } else { + Err(ArtifactError::Storage) + } +} +fn chmod_read_only(fd: i32) -> Result<(), ArtifactError> { + if unsafe { libc::fchmod(fd, 0o400) } == 0 { + Ok(()) + } else { + Err(ArtifactError::Storage) + } +} +fn rename_no_replace(parent: i32, old: &[u8], new: &[u8]) -> Result { + let old = c_name(old)?; + let new = c_name(new)?; + let result = unsafe { + libc::syscall( + libc::SYS_renameat2, + parent, + old.as_ptr(), + parent, + new.as_ptr(), + libc::RENAME_NOREPLACE, + ) + }; + if result == 0 { + Ok(true) + } else if unsafe { *libc::__errno_location() } == libc::EEXIST { + Ok(false) + } else { + Err(ArtifactError::Storage) + } +} +pub(crate) fn unlinkat(parent: i32, name: &[u8]) -> Result<(), ArtifactError> { + let name = c_name(name)?; + if unsafe { libc::unlinkat(parent, name.as_ptr(), 0) } == 0 { + Ok(()) + } else { + Err(classify_errno()) + } +} +pub(crate) fn classify_errno() -> ArtifactError { + match unsafe { *libc::__errno_location() } { + libc::ENOENT => ArtifactError::NotFound, + libc::ELOOP | libc::ENOTDIR => ArtifactError::UnsafeRoot, + _ => ArtifactError::Storage, + } +} +pub(crate) struct RootLock { + fd: OwnedFd, +} +impl RootLock { + pub(crate) fn shared(root: &Root) -> Result { + let fd = duplicate_root(root)?.fd; + if unsafe { libc::flock(fd.as_raw_fd(), libc::LOCK_SH) } == 0 { + Ok(Self { fd }) + } else { + Err(ArtifactError::Storage) + } + } + pub(crate) fn exclusive(root: &Root) -> Result { + let fd = duplicate_root(root)?.fd; + if unsafe { libc::flock(fd.as_raw_fd(), libc::LOCK_EX) } == 0 { + Ok(Self { fd }) + } else { + Err(ArtifactError::Storage) + } + } +} +impl Drop for RootLock { + fn drop(&mut self) { + unsafe { + libc::flock(self.fd.as_raw_fd(), libc::LOCK_UN); + } + } +} diff --git a/crates/crank-artifacts/src/test_support.rs b/crates/crank-artifacts/src/test_support.rs new file mode 100644 index 0000000..6f2342b --- /dev/null +++ b/crates/crank-artifacts/src/test_support.rs @@ -0,0 +1,81 @@ +//! Test-only crash and fault control for syscall-coupled storage checkpoints. + +use std::{ + sync::{Condvar, Mutex, OnceLock}, + time::Duration, +}; + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum FaultAction { + Exit, + Fail, + Hold, +} + +struct State { + checkpoint: Option<(String, FaultAction)>, + held: bool, +} +fn slot() -> &'static (Mutex, Condvar) { + static SLOT: OnceLock<(Mutex, Condvar)> = OnceLock::new(); + SLOT.get_or_init(|| { + ( + Mutex::new(State { + checkpoint: None, + held: false, + }), + Condvar::new(), + ) + }) +} + +pub fn set_checkpoint(stage: impl Into, action: FaultAction) { + let (lock, _) = slot(); + lock.lock() + .expect("artifact test checkpoint lock poisoned") + .checkpoint = Some((stage.into(), action)); +} + +pub fn clear_checkpoint() { + let (lock, wake) = slot(); + let mut state = lock.lock().expect("artifact test checkpoint lock poisoned"); + state.checkpoint = None; + state.held = false; + wake.notify_all(); +} +pub fn wait_until_held(timeout: Duration) -> bool { + let (lock, wake) = slot(); + let state = lock.lock().expect("artifact test checkpoint lock poisoned"); + let (state, _) = wake + .wait_timeout_while(state, timeout, |state| !state.held) + .expect("artifact test checkpoint lock poisoned"); + state.held +} + +pub(crate) fn checkpoint(stage: &str) -> Option { + let (lock, wake) = slot(); + let mut state = lock.lock().expect("artifact test checkpoint lock poisoned"); + let action = state + .checkpoint + .as_ref() + .and_then(|(expected, action)| (expected == stage).then_some(*action)); + if action == Some(FaultAction::Hold) { + state.held = true; + wake.notify_all(); + while state.checkpoint.is_some() { + state = wake + .wait(state) + .expect("artifact test checkpoint lock poisoned"); + } + } + action +} + +pub(crate) fn action(stage: &str) -> Option { + let (lock, _) = slot(); + let state = lock.lock().expect("artifact test checkpoint lock poisoned"); + state + .checkpoint + .as_ref() + .and_then(|(expected, action)| (expected == stage).then_some(*action)) +} diff --git a/crates/crank-artifacts/tests/artifact_store.rs b/crates/crank-artifacts/tests/artifact_store.rs new file mode 100644 index 0000000..eac15e4 --- /dev/null +++ b/crates/crank-artifacts/tests/artifact_store.rs @@ -0,0 +1,980 @@ +use std::{ + fs, + io::{self, Read}, + os::unix::fs::PermissionsExt, + path::PathBuf, + process::{Child, Command, ExitStatus}, + sync::atomic::{AtomicU64, Ordering}, + thread, + time::Duration, +}; + +#[cfg(debug_assertions)] +use crank_artifacts::test_support::{ + FaultAction, clear_checkpoint, set_checkpoint, wait_until_held, +}; +use crank_artifacts::{ArtifactError, ArtifactRef, ArtifactStore, MAX_ARTIFACT_BYTES, StaleTemp}; +use sha2::{Digest, Sha256}; +#[cfg(debug_assertions)] +use std::sync::{Mutex, OnceLock}; + +static NEXT_ROOT: AtomicU64 = AtomicU64::new(0); +#[cfg(debug_assertions)] +fn fault_guard() -> std::sync::MutexGuard<'static, ()> { + static GUARD: OnceLock> = OnceLock::new(); + GUARD.get_or_init(|| Mutex::new(())).lock().unwrap() +} + +struct TestRoot(PathBuf); + +impl TestRoot { + fn new(name: &str) -> Self { + let path = std::env::temp_dir().join(format!( + "crank-artifacts-{name}-{}-{}", + std::process::id(), + NEXT_ROOT.fetch_add(1, Ordering::Relaxed) + )); + fs::create_dir(&path).unwrap(); + fs::set_permissions(&path, fs::Permissions::from_mode(0o700)).unwrap(); + Self(path) + } +} + +impl Drop for TestRoot { + fn drop(&mut self) { + let _ = fs::set_permissions(&self.0, fs::Permissions::from_mode(0o700)); + let _ = fs::remove_dir_all(&self.0); + } +} + +fn wait_for_child(mut child: Child, timeout: Duration) -> ExitStatus { + let deadline = std::time::Instant::now() + timeout; + loop { + if let Some(status) = child.try_wait().unwrap() { + return status; + } + if std::time::Instant::now() >= deadline { + let _ = child.kill(); + let _ = child.wait(); + panic!("child process did not finish within {timeout:?}"); + } + thread::sleep(Duration::from_millis(10)); + } +} + +#[test] +fn stores_and_reads_canonical_digest() { + let root = TestRoot::new("roundtrip"); + let store = ArtifactStore::open(&root.0).unwrap(); + let stored = store.put(b"openapi: 3.0.0\n").unwrap(); + + assert_eq!( + stored.artifact_ref.as_str(), + "sha256:344e4b2f7f15b76b5606be45d8031fc43f473f8d63f0e02c61dbf89a97f85e69" + ); + assert_eq!( + store.read(&stored.artifact_ref).unwrap(), + b"openapi: 3.0.0\n" + ); +} + +#[test] +fn concurrent_writers_deduplicate_without_overwrite() { + let root = TestRoot::new("race"); + let store = ArtifactStore::open(&root.0).unwrap(); + let writers = (0..8) + .map(|_| { + let store = store.clone(); + thread::spawn(move || store.put(b"same immutable bytes").unwrap().artifact_ref) + }) + .collect::>(); + let refs = writers + .into_iter() + .map(|writer| writer.join().unwrap()) + .collect::>(); + + assert!(refs.windows(2).all(|pair| pair[0] == pair[1])); + assert_eq!(store.read(&refs[0]).unwrap(), b"same immutable bytes"); +} + +#[test] +fn independent_writer_processes_deduplicate_without_overwrite() { + if let (Some(root), Some(ready), Some(index)) = ( + std::env::var_os("CRANK_ARTIFACTS_PROCESS_ROOT"), + std::env::var_os("CRANK_ARTIFACTS_PROCESS_READY"), + std::env::var_os("CRANK_ARTIFACTS_PROCESS_INDEX"), + ) { + let ready = PathBuf::from(ready); + fs::write(ready.join(index), b"ready").unwrap(); + let start = ready.join("start"); + while !start.exists() { + thread::sleep(Duration::from_millis(5)); + } + let store = ArtifactStore::open(PathBuf::from(root)).unwrap(); + let stored = store.put(b"same process bytes").unwrap(); + let expected = + ArtifactRef::from_digest_hex(&format!("{:x}", Sha256::digest(b"same process bytes"))) + .unwrap(); + assert_eq!(stored.artifact_ref, expected); + return; + } + + let root = TestRoot::new("process-race"); + let barrier = TestRoot::new("process-barrier"); + let executable = std::env::current_exe().unwrap(); + let mut children = (0..4) + .map(|index| { + Command::new(&executable) + .args([ + "--exact", + "independent_writer_processes_deduplicate_without_overwrite", + "--nocapture", + ]) + .env("CRANK_ARTIFACTS_PROCESS_ROOT", &root.0) + .env("CRANK_ARTIFACTS_PROCESS_READY", &barrier.0) + .env("CRANK_ARTIFACTS_PROCESS_INDEX", index.to_string()) + .spawn() + .unwrap() + }) + .collect::>(); + let deadline = std::time::Instant::now() + Duration::from_secs(3); + while fs::read_dir(&barrier.0).unwrap().count() != children.len() { + assert!( + std::time::Instant::now() < deadline, + "children missed barrier" + ); + thread::sleep(Duration::from_millis(10)); + } + fs::write(barrier.0.join("start"), b"start").unwrap(); + for child in children.drain(..) { + assert!(wait_for_child(child, Duration::from_secs(3)).success()); + } + + let store = ArtifactStore::open(&root.0).unwrap(); + let expected = + ArtifactRef::from_digest_hex(&format!("{:x}", Sha256::digest(b"same process bytes"))) + .unwrap(); + assert_eq!(store.read(&expected).unwrap(), b"same process bytes"); +} + +#[test] +#[cfg(debug_assertions)] +fn verified_dedupe_fsyncs_its_shard_before_reporting_success() { + let _guard = fault_guard(); + let root = TestRoot::new("dedupe-fsync"); + let store = ArtifactStore::open(&root.0).unwrap(); + store.put(b"dedupe fsync bytes").unwrap(); + set_checkpoint("dedupe_directory_fsync", FaultAction::Fail); + assert_eq!( + store.put(b"dedupe fsync bytes").unwrap_err(), + ArtifactError::Storage + ); + clear_checkpoint(); + assert_eq!( + store.put(b"dedupe fsync bytes").unwrap().size_bytes, + b"dedupe fsync bytes".len() + ); +} + +#[test] +#[cfg(debug_assertions)] +fn new_namespace_edges_require_parent_directory_fsync() { + let _guard = fault_guard(); + let root = TestRoot::new("namespace-fsync"); + let store = ArtifactStore::open(&root.0).unwrap(); + set_checkpoint("mkdir_parent_fsync", FaultAction::Fail); + + assert_eq!(store.put(b"namespace bytes"), Err(ArtifactError::Storage)); + assert!(root.0.join("sha256").is_dir()); + clear_checkpoint(); + set_checkpoint("mkdir_parent_fsync", FaultAction::Hold); + let retry_store = store.clone(); + let retry = thread::spawn(move || retry_store.put(b"namespace bytes")); + assert!(wait_until_held(Duration::from_secs(2))); + clear_checkpoint(); + let stored = retry.join().unwrap().unwrap(); + assert_eq!( + store.read(&stored.artifact_ref).unwrap(), + b"namespace bytes" + ); +} + +#[test] +#[cfg(debug_assertions)] +fn live_temp_cannot_be_deleted_while_writer_holds_shared_root_lock() { + let _guard = fault_guard(); + use std::sync::mpsc; + set_checkpoint("write", FaultAction::Hold); + let root = TestRoot::new("live-temp"); + let store = ArtifactStore::open(&root.0).unwrap(); + let writer_store = store.clone(); + let writer = thread::spawn(move || writer_store.put(b"live temp bytes")); + assert!(wait_until_held(Duration::from_secs(2))); + let (_, candidates) = store.scan_stale_temps(Duration::ZERO, 128, 16).unwrap(); + assert_eq!(candidates.len(), 1); + let candidate = candidates.into_iter().next().unwrap(); + let deleter_store = store.clone(); + let (sent, received) = mpsc::channel(); + let deleter = thread::spawn(move || { + let _ = sent.send(deleter_store.delete_stale_temp(candidate)); + }); + assert!(received.recv_timeout(Duration::from_millis(100)).is_err()); + clear_checkpoint(); + assert!(writer.join().unwrap().is_ok()); + assert!(received.recv_timeout(Duration::from_secs(2)).is_ok()); + deleter.join().unwrap(); +} + +#[test] +fn rejects_malformed_and_oversized_inputs() { + assert!(ArtifactRef::parse(&format!("sha256:{}", "a".repeat(64))).is_ok()); + assert!(ArtifactRef::parse(&format!("sha256:{}", "A".repeat(64))).is_err()); + assert!(ArtifactRef::parse("sha256:ABC").is_err()); + assert!(ArtifactRef::parse("file:///tmp/a").is_err()); + let root = TestRoot::new("limits"); + let store = ArtifactStore::open(&root.0).unwrap(); + assert_eq!(store.put(b"").unwrap_err(), ArtifactError::EmptySource); + assert_eq!( + store.put(&vec![0; MAX_ARTIFACT_BYTES + 1]).unwrap_err(), + ArtifactError::SourceTooLarge + ); + let maximum = vec![b'x'; MAX_ARTIFACT_BYTES]; + let stored = store.put(&maximum).unwrap(); + assert_eq!(stored.size_bytes, MAX_ARTIFACT_BYTES); + assert_eq!(store.read(&stored.artifact_ref).unwrap(), maximum); +} + +#[test] +fn root_replacement_cannot_redirect_an_open_store() { + let root = TestRoot::new("replacement"); + let store = ArtifactStore::open(&root.0).unwrap(); + let original = root.0.with_extension("original"); + fs::rename(&root.0, &original).unwrap(); + fs::create_dir(&root.0).unwrap(); + fs::set_permissions(&root.0, fs::Permissions::from_mode(0o700)).unwrap(); + // Replacing the configured pathname after opening cannot change the fd + // authority; only the original inode receives the blob. + store.put(b"pinned authority").unwrap(); + assert!(original.join("sha256").exists()); + assert!(!root.0.join("sha256").exists()); + let _ = fs::remove_dir_all(&original); +} + +#[test] +fn stale_temp_cleanup_uses_a_bounded_capability() { + let root = TestRoot::new("cleanup"); + let store = ArtifactStore::open(&root.0).unwrap(); + let stored = store.put(b"prepare shard").unwrap(); + let shard = root + .0 + .join("sha256") + .join(&stored.artifact_ref.digest_hex()[..2]); + let temp = shard.join(".crank-artifact-tmp-v1-00000000000000000000000000000000-1-1"); + fs::write(&temp, b"partial").unwrap(); + let (report, candidates) = store.scan_stale_temps(Duration::ZERO, 16, 1).unwrap(); + assert!(report.scanned <= 16); + assert_eq!(candidates.len(), 1); + store + .delete_stale_temp(candidates.into_iter().next().unwrap()) + .unwrap(); + assert!(!temp.exists()); +} + +#[test] +fn housekeeping_deletes_entries_older_than_positive_grace() { + let root = TestRoot::new("cleanup-positive-grace"); + let store = ArtifactStore::open(&root.0).unwrap(); + let stored = store.put(b"positive grace shard").unwrap(); + let temp = root + .0 + .join("sha256") + .join(&stored.artifact_ref.digest_hex()[..2]) + .join(".crank-artifact-tmp-v1-00000000000000000000000000000000-1-9"); + fs::write(&temp, b"partial").unwrap(); + fs::File::options() + .write(true) + .open(&temp) + .unwrap() + .set_modified(std::time::SystemTime::now() - Duration::from_secs(5)) + .unwrap(); + + let (_, candidates) = store + .scan_stale_temps(Duration::from_secs(1), 128, 1) + .unwrap(); + assert_eq!(candidates.len(), 1); + store + .delete_stale_temp(candidates.into_iter().next().unwrap()) + .unwrap(); + assert!(!temp.exists()); +} + +#[test] +#[cfg(debug_assertions)] +fn housekeeping_ignores_temp_that_disappears_after_readdir() { + let _guard = fault_guard(); + let root = TestRoot::new("cleanup-disappearing"); + let store = ArtifactStore::open(&root.0).unwrap(); + let stored = store.put(b"disappearing shard").unwrap(); + let temp = root + .0 + .join("sha256") + .join(&stored.artifact_ref.digest_hex()[..2]) + .join(".crank-artifact-tmp-v1-00000000000000000000000000000000-1-11"); + fs::write(&temp, b"partial").unwrap(); + set_checkpoint("housekeeping_before_stat", FaultAction::Hold); + let scan_store = store.clone(); + let scanner = thread::spawn(move || scan_store.scan_stale_temps(Duration::ZERO, 128, 16)); + assert!(wait_until_held(Duration::from_secs(2))); + fs::remove_file(&temp).unwrap(); + clear_checkpoint(); + + let (report, candidates) = scanner.join().unwrap().unwrap(); + assert!(report.scanned > 0); + assert!(candidates.is_empty()); +} + +#[test] +fn housekeeping_enforces_result_limit_after_stale_validation() { + let root = TestRoot::new("cleanup-result-limit"); + let store = ArtifactStore::open(&root.0).unwrap(); + let mut temp_paths = Vec::new(); + for (source, sequence) in [(b"limit-a".as_slice(), 1), (b"limit-b", 2)] { + let stored = store.put(source).unwrap(); + let shard = root + .0 + .join("sha256") + .join(&stored.artifact_ref.digest_hex()[..2]); + let temp = shard.join(format!( + ".crank-artifact-tmp-v1-00000000000000000000000000000000-1-{sequence}" + )); + fs::write(&temp, b"partial").unwrap(); + temp_paths.push(temp); + } + + let (limited, candidates) = store.scan_stale_temps(Duration::ZERO, 128, 1).unwrap(); + assert_eq!(candidates.len(), 1); + assert_eq!(limited.omitted, 1); + let (zero, candidates) = store.scan_stale_temps(Duration::ZERO, 128, 0).unwrap(); + assert!(candidates.is_empty()); + assert_eq!(zero.omitted, 2); + assert!(temp_paths.iter().all(|path| path.exists())); +} + +#[test] +fn housekeeping_revalidates_subsecond_mtime() { + let root = TestRoot::new("cleanup-subsecond"); + let store = ArtifactStore::open(&root.0).unwrap(); + let stored = store.put(b"subsecond shard").unwrap(); + let temp = root + .0 + .join("sha256") + .join(&stored.artifact_ref.digest_hex()[..2]) + .join(".crank-artifact-tmp-v1-00000000000000000000000000000000-1-3"); + fs::write(&temp, b"partial").unwrap(); + let (_, candidates) = store.scan_stale_temps(Duration::ZERO, 128, 1).unwrap(); + assert_eq!(candidates.len(), 1); + + let original = fs::metadata(&temp) + .unwrap() + .modified() + .unwrap() + .duration_since(std::time::UNIX_EPOCH) + .unwrap(); + let changed_nanos = if original.subsec_nanos() == 999_999_999 { + 999_999_998 + } else { + original.subsec_nanos() + 1 + }; + fs::File::options() + .write(true) + .open(&temp) + .unwrap() + .set_modified(std::time::UNIX_EPOCH + Duration::new(original.as_secs(), changed_nanos)) + .unwrap(); + assert_eq!( + store.delete_stale_temp(candidates.into_iter().next().unwrap()), + Err(ArtifactError::UnsafeRoot) + ); + assert!(temp.exists()); +} + +#[test] +fn housekeeping_respects_global_budgets_and_rejects_foreign_or_replaced_temps() { + let root = TestRoot::new("housekeeping-audit"); + let store = ArtifactStore::open(&root.0).unwrap(); + for source in [b"a".as_slice(), b"b", b"c"] { + let _ = store.put(source).unwrap(); + } + let sha = root.0.join("sha256"); + let mut shards = fs::read_dir(&sha) + .unwrap() + .map(|entry| entry.unwrap().path()) + .collect::>(); + shards.sort(); + for shard in &shards { + fs::write(shard.join(".crank-artifact-tmp-v1-not-a-valid-name"), b"x").unwrap(); + } + let replacement = shards[0].join(".crank-artifact-tmp-v1-00000000000000000000000000000000-1-1"); + fs::write(&replacement, b"x").unwrap(); + let (limited, candidates) = store.scan_stale_temps(Duration::ZERO, 5, 1).unwrap(); + assert_eq!(limited.scanned, 5); + assert!(candidates.len() <= 1); + let (_, candidates) = store.scan_stale_temps(Duration::ZERO, 128, 16).unwrap(); + assert_eq!(candidates.len(), 1); + let candidate = candidates.into_iter().next().unwrap(); + fs::remove_file(&replacement).unwrap(); + fs::write(&replacement, b"replacement").unwrap(); + assert_eq!( + store.delete_stale_temp(candidate).unwrap_err(), + ArtifactError::UnsafeRoot + ); + let (_, candidates) = store.scan_stale_temps(Duration::ZERO, 128, 16).unwrap(); + let foreign_candidate = candidates.into_iter().next().unwrap(); + let other = TestRoot::new("housekeeping-other"); + let other_store = ArtifactStore::open(&other.0).unwrap(); + assert_eq!( + other_store + .delete_stale_temp(foreign_candidate) + .unwrap_err(), + ArtifactError::UnsafeRoot + ); + let (_, fresh) = store + .scan_stale_temps(Duration::from_secs(3600), 128, 16) + .unwrap(); + assert!(fresh.is_empty()); +} + +#[test] +fn open_rejects_non_private_root_and_existing_finals_must_be_immutable() { + let root = TestRoot::new("root-audit"); + fs::set_permissions(&root.0, fs::Permissions::from_mode(0o755)).unwrap(); + assert_eq!( + ArtifactStore::open(&root.0).unwrap_err(), + ArtifactError::UnsafeRoot + ); + fs::set_permissions(&root.0, fs::Permissions::from_mode(0o770)).unwrap(); + assert_eq!( + ArtifactStore::open(&root.0).unwrap_err(), + ArtifactError::UnsafeRoot + ); + fs::set_permissions(&root.0, fs::Permissions::from_mode(0o700)).unwrap(); + let store = ArtifactStore::open(&root.0).unwrap(); + let stored = store.put(b"final audit").unwrap(); + let path = root + .0 + .join("sha256") + .join(&stored.artifact_ref.digest_hex()[..2]) + .join(stored.artifact_ref.digest_hex()); + fs::set_permissions(&path, fs::Permissions::from_mode(0o600)).unwrap(); + assert_eq!( + store.read(&stored.artifact_ref).unwrap_err(), + ArtifactError::Integrity + ); + fs::set_permissions(&path, fs::Permissions::from_mode(0o400)).unwrap(); + let linked = path.with_extension("link"); + fs::hard_link(&path, &linked).unwrap(); + assert_eq!( + store.read(&stored.artifact_ref).unwrap_err(), + ArtifactError::Integrity + ); +} + +#[test] +fn open_rejects_symlinked_root() { + let root = TestRoot::new("root-symlink-target"); + let link = root.0.with_extension("symlink"); + std::os::unix::fs::symlink(&root.0, &link).unwrap(); + assert!(ArtifactStore::open(&link).is_err()); + fs::remove_file(link).unwrap(); +} + +#[test] +fn corrupted_or_symlinked_final_never_returns_bytes() { + let root = TestRoot::new("integrity"); + let store = ArtifactStore::open(&root.0).unwrap(); + let stored = store.put(b"safe bytes").unwrap(); + let path = root + .0 + .join("sha256") + .join(&stored.artifact_ref.digest_hex()[..2]) + .join(stored.artifact_ref.digest_hex()); + fs::set_permissions(&path, fs::Permissions::from_mode(0o600)).unwrap(); + fs::write(&path, b"evil bytes").unwrap(); + fs::set_permissions(&path, fs::Permissions::from_mode(0o400)).unwrap(); + assert_eq!( + store.read(&stored.artifact_ref).unwrap_err(), + ArtifactError::Integrity + ); + + fs::remove_file(&path).unwrap(); + let outside = root.0.join("outside-read-only"); + fs::write(&outside, b"safe bytes").unwrap(); + fs::set_permissions(&outside, fs::Permissions::from_mode(0o400)).unwrap(); + std::os::unix::fs::symlink(&outside, &path).unwrap(); + assert!(store.read(&stored.artifact_ref).is_err()); + + fs::remove_file(&path).unwrap(); + fs::create_dir(&path).unwrap(); + assert_eq!( + store.read(&stored.artifact_ref).unwrap_err(), + ArtifactError::Integrity + ); +} + +#[test] +fn directory_at_final_digest_is_integrity_failure() { + let root = TestRoot::new("directory-final"); + let store = ArtifactStore::open(&root.0).unwrap(); + let stored = store.put(b"directory final").unwrap(); + let path = root + .0 + .join("sha256") + .join(&stored.artifact_ref.digest_hex()[..2]) + .join(stored.artifact_ref.digest_hex()); + fs::remove_file(&path).unwrap(); + fs::create_dir(&path).unwrap(); + fs::set_permissions(&path, fs::Permissions::from_mode(0o700)).unwrap(); + assert_eq!( + store.read(&stored.artifact_ref).unwrap_err(), + ArtifactError::Integrity + ); +} + +#[test] +fn legacy_reader_is_pinned_to_its_explicit_root_and_digest() { + let root = TestRoot::new("legacy"); + let legacy = root.0.join("legacy"); + fs::create_dir(&legacy).unwrap(); + fs::set_permissions(&legacy, fs::Permissions::from_mode(0o700)).unwrap(); + let source = legacy.join("source.yaml"); + fs::write(&source, b"openapi: 3.0.0").unwrap(); + fs::set_permissions(&source, fs::Permissions::from_mode(0o600)).unwrap(); + let expected = + ArtifactRef::from_digest_hex(&format!("{:x}", Sha256::digest(b"openapi: 3.0.0"))).unwrap(); + let store = ArtifactStore::open(&root.0) + .unwrap() + .with_legacy_root(&legacy); + + assert_eq!( + store + .read_legacy_file_url(&format!("file://{}", source.display()), &expected) + .unwrap(), + b"openapi: 3.0.0" + ); + assert!(matches!( + store.read_legacy_file_url("file:///etc/passwd", &expected), + Err(ArtifactError::UnsafeRoot) + )); + assert_eq!( + store + .read_legacy_file_url( + &format!("file://{}", source.display()), + &ArtifactRef::from_digest_hex(&"0".repeat(64)).unwrap() + ) + .unwrap_err(), + ArtifactError::Integrity + ); + let outside = root.0.join("outside-legacy.yaml"); + fs::write(&outside, b"openapi: 3.0.0").unwrap(); + fs::remove_file(&source).unwrap(); + std::os::unix::fs::symlink(&outside, &source).unwrap(); + assert!( + store + .read_legacy_file_url(&format!("file://{}", source.display()), &expected) + .is_err() + ); + + let maximum = legacy.join("maximum.bin"); + let maximum_bytes = vec![b'm'; MAX_ARTIFACT_BYTES]; + fs::write(&maximum, &maximum_bytes).unwrap(); + let maximum_ref = + ArtifactRef::from_digest_hex(&format!("{:x}", Sha256::digest(&maximum_bytes))).unwrap(); + assert_eq!( + store + .read_legacy_file_url(&format!("file://{}", maximum.display()), &maximum_ref) + .unwrap(), + maximum_bytes + ); + assert!(maximum.exists()); +} + +#[test] +fn legacy_reader_handles_lexical_parent_components_without_following_symlinks() { + let root = TestRoot::new("legacy-normalized"); + let base = root.0.join("base"); + let child = base.join("child"); + let legacy = base.join("legacy"); + fs::create_dir_all(&child).unwrap(); + fs::create_dir(&legacy).unwrap(); + fs::set_permissions(&base, fs::Permissions::from_mode(0o700)).unwrap(); + fs::set_permissions(&child, fs::Permissions::from_mode(0o700)).unwrap(); + fs::set_permissions(&legacy, fs::Permissions::from_mode(0o700)).unwrap(); + let source = legacy.join("source.yaml"); + fs::write(&source, b"normalized legacy").unwrap(); + let expected = + ArtifactRef::from_digest_hex(&format!("{:x}", Sha256::digest(b"normalized legacy"))) + .unwrap(); + let configured = child.join("..").join("legacy"); + let store = ArtifactStore::open(&root.0) + .unwrap() + .with_legacy_root(configured); + + assert_eq!( + store + .read_legacy_file_url(&format!("file://{}", source.display()), &expected) + .unwrap(), + b"normalized legacy" + ); + + let outside_dir = base.join("outside"); + fs::create_dir(&outside_dir).unwrap(); + fs::write(outside_dir.join("source.yaml"), b"normalized legacy").unwrap(); + std::os::unix::fs::symlink(&outside_dir, legacy.join("linked")).unwrap(); + assert!( + store + .read_legacy_file_url( + &format!("file://{}/linked/source.yaml", legacy.display()), + &expected, + ) + .is_err() + ); +} + +#[test] +fn legacy_oversize_is_rejected() { + let root = TestRoot::new("legacy-max"); + let legacy = root.0.join("legacy"); + fs::create_dir(&legacy).unwrap(); + fs::set_permissions(&legacy, fs::Permissions::from_mode(0o700)).unwrap(); + let source = legacy.join("large.yaml"); + fs::write(&source, vec![b'x'; MAX_ARTIFACT_BYTES + 1]).unwrap(); + let expected = ArtifactRef::from_digest_hex(&format!( + "{:x}", + Sha256::digest(vec![b'x'; MAX_ARTIFACT_BYTES + 1]) + )) + .unwrap(); + let store = ArtifactStore::open(&root.0) + .unwrap() + .with_legacy_root(&legacy); + assert_eq!( + store + .read_legacy_file_url(&format!("file://{}", source.display()), &expected) + .unwrap_err(), + ArtifactError::Integrity + ); +} + +#[test] +fn fifo_entries_fail_without_blocking_reads() { + let root = TestRoot::new("fifo"); + let expected = + ArtifactRef::from_digest_hex(&format!("{:x}", Sha256::digest(b"fifo bytes"))).unwrap(); + let sha = root.0.join("sha256"); + let shard = sha.join(&expected.digest_hex()[..2]); + fs::create_dir(&sha).unwrap(); + fs::create_dir(&shard).unwrap(); + fs::set_permissions(&sha, fs::Permissions::from_mode(0o700)).unwrap(); + fs::set_permissions(&shard, fs::Permissions::from_mode(0o700)).unwrap(); + let blob_fifo = shard.join(expected.digest_hex()); + let legacy = root.0.join("legacy"); + fs::create_dir(&legacy).unwrap(); + fs::set_permissions(&legacy, fs::Permissions::from_mode(0o700)).unwrap(); + let legacy_fifo = legacy.join("source"); + for path in [&blob_fifo, &legacy_fifo] { + let path = std::ffi::CString::new(path.as_os_str().as_encoded_bytes()).unwrap(); + assert_eq!(unsafe { libc::mkfifo(path.as_ptr(), 0o400) }, 0); + } + + let executable = std::env::current_exe().unwrap(); + for kind in ["blob", "legacy"] { + let child = Command::new(&executable) + .args(["--exact", "fifo_read_child", "--nocapture"]) + .env("CRANK_ARTIFACTS_FIFO_ROOT", &root.0) + .env("CRANK_ARTIFACTS_FIFO_KIND", kind) + .env("CRANK_ARTIFACTS_FIFO_REF", expected.as_str()) + .spawn() + .unwrap(); + assert!(wait_for_child(child, Duration::from_secs(2)).success()); + } +} + +#[test] +fn fifo_read_child() { + let Some(root) = std::env::var_os("CRANK_ARTIFACTS_FIFO_ROOT") else { + return; + }; + let root = PathBuf::from(root); + let kind = std::env::var("CRANK_ARTIFACTS_FIFO_KIND").unwrap(); + let expected = ArtifactRef::parse(&std::env::var("CRANK_ARTIFACTS_FIFO_REF").unwrap()).unwrap(); + let store = ArtifactStore::open(&root).unwrap(); + let result = if kind == "blob" { + store.read(&expected) + } else { + store + .with_legacy_root(root.join("legacy")) + .read_legacy_file_url( + &format!("file://{}/legacy/source", root.display()), + &expected, + ) + }; + assert_eq!(result.unwrap_err(), ArtifactError::Integrity); +} + +struct ShortReader { + bytes: Vec, + offset: usize, + interrupted: bool, +} + +struct FailingReader; + +struct HousekeepingReader { + store: ArtifactStore, + candidate: Option, + bytes: io::Cursor>, +} + +impl Read for FailingReader { + fn read(&mut self, _buf: &mut [u8]) -> io::Result { + Err(io::Error::other("private reader detail")) + } +} +impl Read for HousekeepingReader { + fn read(&mut self, buffer: &mut [u8]) -> io::Result { + if let Some(candidate) = self.candidate.take() { + self.store.delete_stale_temp(candidate).unwrap(); + } + self.bytes.read(buffer) + } +} +impl Read for ShortReader { + fn read(&mut self, buf: &mut [u8]) -> io::Result { + if !self.interrupted { + self.interrupted = true; + return Err(io::Error::from(io::ErrorKind::Interrupted)); + } + if self.offset == self.bytes.len() { + return Ok(0); + } + let amount = 1.min(buf.len()).min(self.bytes.len() - self.offset); + buf[..amount].copy_from_slice(&self.bytes[self.offset..self.offset + amount]); + self.offset += amount; + Ok(amount) + } +} + +#[test] +fn streaming_input_handles_short_reads_and_interruption() { + let root = TestRoot::new("stream"); + let store = ArtifactStore::open(&root.0).unwrap(); + let mut reader = ShortReader { + bytes: b"short chunk source".to_vec(), + offset: 0, + interrupted: false, + }; + let stored = store.put_reader(&mut reader).unwrap(); + assert_eq!( + store.read(&stored.artifact_ref).unwrap(), + b"short chunk source" + ); + + assert_eq!( + store.put_reader(&mut FailingReader).unwrap_err(), + ArtifactError::Storage + ); + for error in [ + ArtifactError::InvalidReference, + ArtifactError::Integrity, + ArtifactError::Storage, + ArtifactError::UnsafeRoot, + ] { + let diagnostic = format!("{error} {error:?}"); + assert!(!diagnostic.contains("private reader detail")); + assert!(!diagnostic.contains(std::env::temp_dir().to_string_lossy().as_ref())); + } +} + +#[test] +fn user_reader_can_run_housekeeping_without_self_deadlock() { + use std::sync::mpsc; + + let root = TestRoot::new("reader-housekeeping"); + let store = ArtifactStore::open(&root.0).unwrap(); + let stored = store.put(b"reader housekeeping shard").unwrap(); + let temp = root + .0 + .join("sha256") + .join(&stored.artifact_ref.digest_hex()[..2]) + .join(".crank-artifact-tmp-v1-00000000000000000000000000000000-1-10"); + fs::write(&temp, b"partial").unwrap(); + let (_, candidates) = store.scan_stale_temps(Duration::ZERO, 128, 1).unwrap(); + let mut reader = HousekeepingReader { + store: store.clone(), + candidate: candidates.into_iter().next(), + bytes: io::Cursor::new(b"reader bytes".to_vec()), + }; + let writer_store = store.clone(); + let (sender, receiver) = mpsc::channel(); + thread::spawn(move || { + sender.send(writer_store.put_reader(&mut reader)).unwrap(); + }); + + let stored = receiver + .recv_timeout(Duration::from_secs(2)) + .expect("put_reader deadlocked while its reader ran housekeeping") + .unwrap(); + assert_eq!(store.read(&stored.artifact_ref).unwrap(), b"reader bytes"); + assert!(!temp.exists()); +} + +#[test] +#[cfg(debug_assertions)] +fn forked_crash_checkpoints_publish_only_complete_or_absent_blobs() { + let _guard = fault_guard(); + if let Some(root) = std::env::var_os("CRANK_ARTIFACTS_CHILD_ROOT") { + set_checkpoint( + std::env::var("CRANK_ARTIFACTS_TEST_STAGE").unwrap(), + FaultAction::Exit, + ); + let store = ArtifactStore::open(PathBuf::from(root)).unwrap(); + let _ = store.put(b"crash-consistent bytes"); + panic!("checkpoint did not terminate the child"); + } + + for stage in ["write", "file_fsync", "publish", "directory_fsync"] { + let root = TestRoot::new(stage); + let status = Command::new(std::env::current_exe().unwrap()) + .args([ + "--exact", + "forked_crash_checkpoints_publish_only_complete_or_absent_blobs", + "--nocapture", + ]) + .env("CRANK_ARTIFACTS_CHILD_ROOT", &root.0) + .env("CRANK_ARTIFACTS_TEST_STAGE", stage) + .status() + .unwrap(); + assert_eq!(status.code(), Some(86), "stage={stage}"); + + let store = ArtifactStore::open(&root.0).unwrap(); + let expected = ArtifactRef::from_digest_hex(&format!( + "{:x}", + Sha256::digest(b"crash-consistent bytes") + )) + .unwrap(); + match store.read(&expected) { + Ok(bytes) => assert_eq!(bytes, b"crash-consistent bytes", "stage={stage}"), + Err(ArtifactError::NotFound) => {} + Err(error) => panic!("stage={stage} exposed an invalid final blob: {error}"), + } + let stored = store.put(b"crash-consistent bytes").unwrap(); + assert_eq!(stored.artifact_ref, expected, "stage={stage}"); + assert_eq!( + store.read(&expected).unwrap(), + b"crash-consistent bytes", + "stage={stage}" + ); + + let (_, candidates) = store.scan_stale_temps(Duration::ZERO, 128, 16).unwrap(); + for candidate in candidates { + store.delete_stale_temp(candidate).unwrap(); + } + } +} + +#[test] +#[cfg(debug_assertions)] +fn disk_full_fault_seams_leave_a_retryable_store() { + let _guard = fault_guard(); + if let Some(root) = std::env::var_os("CRANK_ARTIFACTS_CHILD_ROOT") { + set_checkpoint( + std::env::var("CRANK_ARTIFACTS_TEST_STAGE").unwrap(), + FaultAction::Fail, + ); + let store = ArtifactStore::open(PathBuf::from(root)).unwrap(); + assert_eq!( + store.put(b"disk-full seam bytes").unwrap_err(), + ArtifactError::Storage + ); + return; + } + + for stage in ["write", "file_fsync", "publish", "directory_fsync"] { + let root = TestRoot::new(&format!("full-{stage}")); + let status = Command::new(std::env::current_exe().unwrap()) + .args([ + "--exact", + "disk_full_fault_seams_leave_a_retryable_store", + "--nocapture", + ]) + .env("CRANK_ARTIFACTS_CHILD_ROOT", &root.0) + .env("CRANK_ARTIFACTS_TEST_STAGE", stage) + .status() + .unwrap(); + assert!(status.success(), "stage={stage}"); + + let store = ArtifactStore::open(&root.0).unwrap(); + let expected = + ArtifactRef::from_digest_hex(&format!("{:x}", Sha256::digest(b"disk-full seam bytes"))) + .unwrap(); + match store.read(&expected) { + Ok(bytes) => assert_eq!(bytes, b"disk-full seam bytes", "stage={stage}"), + Err(ArtifactError::NotFound) => {} + Err(error) => panic!("stage={stage} exposed an invalid final blob: {error}"), + } + assert_eq!( + store.put(b"disk-full seam bytes").unwrap().artifact_ref, + expected, + "stage={stage}" + ); + assert_eq!( + store.read(&expected).unwrap(), + b"disk-full seam bytes", + "stage={stage}" + ); + } +} + +#[test] +fn real_process_write_limit_leaves_no_partial_final_blob() { + if let Some(root) = std::env::var_os("CRANK_ARTIFACTS_CHILD_ROOT") { + // Keep the test's host filesystem untouched while making the actual + // write syscall fail as it would for a quota/disk-full condition. + unsafe { + assert_ne!(libc::signal(libc::SIGXFSZ, libc::SIG_IGN), libc::SIG_ERR); + let limit = libc::rlimit { + rlim_cur: 1, + rlim_max: 1, + }; + assert_eq!(libc::setrlimit(libc::RLIMIT_FSIZE, &limit), 0); + } + let store = ArtifactStore::open(PathBuf::from(root)).unwrap(); + assert_eq!( + store.put(b"real write limit bytes").unwrap_err(), + ArtifactError::Storage + ); + return; + } + + let root = TestRoot::new("real-write-limit"); + let status = Command::new(std::env::current_exe().unwrap()) + .args([ + "--exact", + "real_process_write_limit_leaves_no_partial_final_blob", + "--nocapture", + ]) + .env("CRANK_ARTIFACTS_CHILD_ROOT", &root.0) + .status() + .unwrap(); + assert!(status.success()); + + let store = ArtifactStore::open(&root.0).unwrap(); + let expected = + ArtifactRef::from_digest_hex(&format!("{:x}", Sha256::digest(b"real write limit bytes"))) + .unwrap(); + assert_eq!(store.read(&expected).unwrap_err(), ArtifactError::NotFound); + assert_eq!( + store.put(b"real write limit bytes").unwrap().artifact_ref, + expected + ); + assert_eq!(store.read(&expected).unwrap(), b"real write limit bytes"); +} diff --git a/scripts/check-rust-boundaries.py b/scripts/check-rust-boundaries.py index 748233d..28d33b7 100755 --- a/scripts/check-rust-boundaries.py +++ b/scripts/check-rust-boundaries.py @@ -53,6 +53,8 @@ def package_category(name: str, manifest_path: Path, workspace_root: Path) -> st return "metrics" if name == "crank-config": return "config" + if name == "crank-artifacts": + return "artifacts" if name == "crank-observability": return "observability" if name == "crank-registry": @@ -86,13 +88,22 @@ def workspace_packages(metadata: dict[str, Any]) -> dict[str, Package]: return packages -def dependency_package_ids(raw_package: dict[str, Any], packages_by_name: dict[str, str]) -> list[str]: +def dependency_package_ids( + raw_package: dict[str, Any], + packages_by_name: dict[str, str], + *, + skip_dev: bool = False, +) -> list[str]: dependency_ids: list[str] = [] + seen: set[str] = set() for dependency in raw_package.get("dependencies", []): + if skip_dev and dependency.get("kind") == "dev": + continue dependency_name = dependency["name"] dependency_id = packages_by_name.get(dependency_name) - if dependency_id is not None: + if dependency_id is not None and dependency_id not in seen: dependency_ids.append(dependency_id) + seen.add(dependency_id) return dependency_ids @@ -128,6 +139,9 @@ def boundary_reason(source: Package, dependency: Package) -> str | None: if source.category == "config": return "crank-config must not depend on other workspace crates" + if source.category == "artifacts": + return "crank-artifacts must not depend on other workspace crates" + if source.category == "observability" and dependency.category != "metrics": return "crank-observability must not depend on other workspace crates" @@ -161,7 +175,11 @@ def find_violations(metadata: dict[str, Any]) -> list[Violation]: for source in sorted(packages.values(), key=lambda package: package.name): raw_package = raw_packages_by_id[source.id] violations.extend(direct_dependency_violations(raw_package)) - for dependency_id in dependency_package_ids(raw_package, packages_by_name): + for dependency_id in dependency_package_ids( + raw_package, + packages_by_name, + skip_dev=source.category == "artifacts", + ): dependency = packages[dependency_id] reason = boundary_reason(source, dependency) if reason is not None: diff --git a/tests/unit/test_check_rust_boundaries.py b/tests/unit/test_check_rust_boundaries.py index 5fc3b4f..cf39a9e 100644 --- a/tests/unit/test_check_rust_boundaries.py +++ b/tests/unit/test_check_rust_boundaries.py @@ -183,6 +183,83 @@ class RustBoundaryCheckTests(unittest.TestCase): self.assertEqual(violations[0].source, "crank-config") self.assertEqual(violations[0].dependency, "crank-core") + def test_rejects_artifact_store_dependency_on_workspace_crates(self) -> None: + packages = [ + package( + self.root, + "crank-artifacts", + "crates/crank-artifacts", + ["crank-core"], + ), + package(self.root, "crank-core", "crates/crank-core"), + ] + + violations = self.checker.find_violations(metadata(packages, self.root)) + + self.assertEqual(len(violations), 1) + self.assertEqual(violations[0].source, "crank-artifacts") + self.assertEqual(violations[0].dependency, "crank-core") + + def test_allows_artifact_store_dev_dependency_on_workspace_crates(self) -> None: + artifacts = package( + self.root, + "crank-artifacts", + "crates/crank-artifacts", + ) + artifacts["dependencies"] = [dependency("crank-test-support", "dev")] + packages = [ + artifacts, + package( + self.root, + "crank-test-support", + "crates/crank-test-support", + ), + ] + + violations = self.checker.find_violations(metadata(packages, self.root)) + + self.assertEqual(violations, []) + + def test_rejects_artifact_store_build_dependency_on_workspace_crates(self) -> None: + artifacts = package( + self.root, + "crank-artifacts", + "crates/crank-artifacts", + ) + artifacts["dependencies"] = [dependency("crank-core", "build")] + packages = [ + artifacts, + package(self.root, "crank-core", "crates/crank-core"), + ] + + violations = self.checker.find_violations(metadata(packages, self.root)) + + self.assertEqual(len(violations), 1) + self.assertEqual(violations[0].source, "crank-artifacts") + self.assertEqual(violations[0].dependency, "crank-core") + + def test_mixed_dependency_kinds_produce_one_production_violation(self) -> None: + artifacts = package( + self.root, + "crank-artifacts", + "crates/crank-artifacts", + ) + artifacts["dependencies"] = [ + dependency("crank-core", "dev"), + dependency("crank-core"), + dependency("crank-core", "build"), + ] + packages = [ + artifacts, + package(self.root, "crank-core", "crates/crank-core"), + ] + + violations = self.checker.find_violations(metadata(packages, self.root)) + + self.assertEqual(len(violations), 1) + self.assertEqual(violations[0].source, "crank-artifacts") + self.assertEqual(violations[0].dependency, "crank-core") + def test_rejects_domain_and_runtime_dependencies_on_observability(self) -> None: for source in ("crank-core", "crank-registry", "crank-runtime"): packages = [