From 8b2f1c0ced463b5fe0fcfc8cdc70d323a27e6203 Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Wed, 12 Aug 2026 17:05:26 +0000 Subject: [PATCH] test: cover transfer-helper scan/repair/CLI and agent terminal, reaper, fsevent paths Co-Authored-By: Augustus Otu --- dory-core/agent/src/reaper.rs | 51 ++++ dory-core/agent/src/terminal.rs | 141 +++++++++ dory-core/agent/src/vsock_server.rs | 139 ++++++++- dory-core/transfer-helper/src/lib.rs | 48 +++ dory-core/transfer-helper/src/main.rs | 182 +++++++++++ dory-core/transfer-helper/src/repair_linux.rs | 285 ++++++++++++++++++ dory-core/transfer-helper/src/scan_linux.rs | 281 +++++++++++++++++ dory-core/transfer-helper/src/test_support.rs | 112 +++++++ 8 files changed, 1236 insertions(+), 3 deletions(-) create mode 100644 dory-core/transfer-helper/src/test_support.rs diff --git a/dory-core/agent/src/reaper.rs b/dory-core/agent/src/reaper.rs index edac6800..491c147d 100644 --- a/dory-core/agent/src/reaper.rs +++ b/dory-core/agent/src/reaper.rs @@ -47,3 +47,54 @@ fn reap_available_children() { } } } + +#[cfg(test)] +mod tests { + use super::*; + + #[tokio::test] + async fn the_child_wait_lock_is_shared_and_exclusive() { + let guard = managed_child_wait_guard().await; + // A background reaper sweep only runs while nothing holds the managed wait, so the lock the + // sweep polls must be the very lock exec waits take. + assert!(child_wait_lock().try_lock().is_err()); + assert!(tokio::time::timeout( + std::time::Duration::from_millis(100), + managed_child_wait_guard() + ) + .await + .is_err()); + + drop(guard); + let reacquired = tokio::time::timeout( + std::time::Duration::from_secs(20), + managed_child_wait_guard(), + ) + .await; + assert!(reacquired.is_ok(), "the released lock must be reacquirable"); + } + + #[cfg(target_os = "linux")] + #[test] + fn the_pid1_reaper_does_not_start_outside_pid_1() { + assert_ne!(unsafe { libc::getpid() }, 1, "tests do not run as PID 1"); + // Starting a `waitpid(-1, ...)` sweep in a non-init process would steal exit statuses from + // the managed exec waits, so the entry point must be a no-op here: an unrelated child stays + // waitable by its own spawner. + start_pid1_reaper_if_needed(); + + let child = std::process::Command::new("/bin/sh") + .args(["-c", "exit 3"]) + .spawn() + .expect("spawn child"); + let pid = child.id() as libc::pid_t; + std::mem::forget(child); + std::thread::sleep(std::time::Duration::from_millis(400)); + + let mut status = 0; + let waited = unsafe { libc::waitpid(pid, &mut status, 0) }; + assert_eq!(waited, pid, "the child was reaped by something else"); + assert!(libc::WIFEXITED(status)); + assert_eq!(libc::WEXITSTATUS(status), 3); + } +} diff --git a/dory-core/agent/src/terminal.rs b/dory-core/agent/src/terminal.rs index 0fc623f7..a058d209 100644 --- a/dory-core/agent/src/terminal.rs +++ b/dory-core/agent/src/terminal.rs @@ -227,6 +227,9 @@ fn terminate_child(pid: libc::pid_t) { #[cfg(test)] mod tests { use super::*; + use tokio::net::UnixStream; + + const STREAM_TIMEOUT: Duration = Duration::from_secs(20); #[test] fn selects_existing_shell() { @@ -237,4 +240,142 @@ mod tests { let _ = libc::waitpid(process.pid, &mut status, 0); } } + + #[test] + fn the_child_shell_receives_the_injected_term() { + let shell = spawn_shell().expect("spawn shell"); + set_nonblocking(shell.master.as_raw_fd()).expect("nonblocking master"); + write_blocking( + shell.master.as_raw_fd(), + b"printf 'term=%s\\n' \"$TERM\"\nexit\n", + ); + + let transcript = read_until(shell.master.as_raw_fd(), "term=xterm-256color"); + terminate_child_blocking(shell.pid); + assert!(transcript.contains("term=xterm-256color"), "{transcript}"); + } + + #[tokio::test] + async fn the_stream_carries_shell_output_and_ends_when_the_shell_exits() { + let (mut host, guest) = UnixStream::pair().expect("socket pair"); + let served = tokio::spawn(serve_shell_stream(guest)); + + host.write_all(b"printf 'dory-%s\\n' ready\nexit\n") + .await + .expect("write command"); + host.flush().await.expect("flush command"); + + let mut transcript = Vec::new(); + let mut buffer = [0_u8; 4096]; + let read = tokio::time::timeout(STREAM_TIMEOUT, async { + loop { + let count = host.read(&mut buffer).await?; + if count == 0 { + return Ok::<(), io::Error>(()); + } + transcript.extend_from_slice(&buffer[..count]); + if String::from_utf8_lossy(&transcript).contains("dory-ready") { + return Ok(()); + } + } + }) + .await; + assert!(read.is_ok(), "timed out reading the shell transcript"); + assert!( + String::from_utf8_lossy(&transcript).contains("dory-ready"), + "{}", + String::from_utf8_lossy(&transcript) + ); + + // The shell exited, so the PTY read side reports EOF and both copy loops unwind. + let outcome = tokio::time::timeout(STREAM_TIMEOUT, served).await; + assert!(outcome.expect("shell stream ended").is_ok()); + } + + #[tokio::test] + async fn closing_the_client_stream_tears_the_session_down() { + let (host, guest) = UnixStream::pair().expect("socket pair"); + let served = tokio::spawn(serve_shell_stream(guest)); + drop(host); + + let outcome = tokio::time::timeout(STREAM_TIMEOUT, served).await; + assert!(outcome.expect("shell stream ended").is_ok()); + } + + #[tokio::test] + async fn pty_reads_and_writes_go_through_the_readiness_guards() { + let shell = spawn_shell().expect("spawn shell"); + set_nonblocking(shell.master.as_raw_fd()).expect("nonblocking master"); + let pid = shell.pid; + let pty = AsyncFd::new(shell.master).expect("register pty"); + + write_all_fd(&pty, b"").await.expect("empty write"); + write_all_fd(&pty, b"printf 'guarded'\n") + .await + .expect("write command"); + let mut buffer = [0_u8; 4096]; + let count = tokio::time::timeout(STREAM_TIMEOUT, read_fd(&pty, &mut buffer)) + .await + .expect("read did not time out") + .expect("read"); + assert!(count > 0); + terminate_child_blocking(pid); + } + + #[test] + fn raw_descriptor_helpers_surface_their_errors() { + assert_eq!(dup_fd(-1).unwrap_err().raw_os_error(), Some(libc::EBADF)); + assert_eq!( + set_nonblocking(-1).unwrap_err().raw_os_error(), + Some(libc::EBADF) + ); + assert_eq!( + read_raw(-1, &mut [0_u8; 1]).unwrap_err().raw_os_error(), + Some(libc::EBADF) + ); + assert_eq!( + write_raw(-1, b"x").unwrap_err().raw_os_error(), + Some(libc::EBADF) + ); + } + + fn write_blocking(fd: RawFd, mut bytes: &[u8]) { + while !bytes.is_empty() { + match write_raw(fd, bytes) { + Ok(count) => bytes = &bytes[count..], + Err(error) if error.kind() == io::ErrorKind::WouldBlock => { + std::thread::sleep(Duration::from_millis(20)) + } + Err(error) => panic!("pty write failed: {error}"), + } + } + } + + fn read_until(fd: RawFd, marker: &str) -> String { + let deadline = std::time::Instant::now() + STREAM_TIMEOUT; + let mut output = Vec::new(); + let mut buffer = [0_u8; 4096]; + while std::time::Instant::now() < deadline { + match read_raw(fd, &mut buffer) { + Ok(0) => break, + Ok(count) => output.extend_from_slice(&buffer[..count]), + Err(error) if error.kind() == io::ErrorKind::WouldBlock => { + std::thread::sleep(Duration::from_millis(20)) + } + Err(_) => break, + } + if String::from_utf8_lossy(&output).contains(marker) { + break; + } + } + String::from_utf8_lossy(&output).into_owned() + } + + fn terminate_child_blocking(pid: libc::pid_t) { + unsafe { + libc::kill(pid, libc::SIGKILL); + let mut status = 0; + let _ = libc::waitpid(pid, &mut status, 0); + } + } } diff --git a/dory-core/agent/src/vsock_server.rs b/dory-core/agent/src/vsock_server.rs index 8de60c99..7b8e5e38 100644 --- a/dory-core/agent/src/vsock_server.rs +++ b/dory-core/agent/src/vsock_server.rs @@ -112,10 +112,13 @@ async fn serve_fsevents() -> std::io::Result<()> { } } -async fn handle_fsevent_batch( - mut stream: tokio_vsock::VsockStream, +async fn handle_fsevent_batch( + mut stream: S, dedupe: Arc, -) -> std::io::Result<()> { +) -> std::io::Result<()> +where + S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin, +{ // One deadline covers the complete frame so a peer cannot slow-drip a permitted connection. let body = timeout(FSEVENT_READ_TIMEOUT, async { let body_len = stream.read_u32_le().await? as usize; @@ -241,10 +244,140 @@ async fn serve_shell() -> std::io::Result<()> { mod tests { use super::*; + use std::path::PathBuf; + + use crate::fsevents::{encode_batch_frame, ResponseStatus, MAX_FRAME_BYTES, PROTOCOL_VERSION}; + + struct Response { + operation_id: u64, + path_count: u32, + status: u32, + failed_indices: Vec, + } + + fn decode_response(frame: &[u8]) -> Response { + let body_len = u32::from_le_bytes(frame[0..4].try_into().unwrap()) as usize; + assert_eq!(frame.len(), 4 + body_len); + assert_eq!( + u32::from_le_bytes(frame[4..8].try_into().unwrap()), + PROTOCOL_VERSION + ); + let failed_count = u32::from_le_bytes(frame[24..28].try_into().unwrap()) as usize; + let failed_indices = (0..failed_count) + .map(|index| { + let start = 28 + index * 4; + u32::from_le_bytes(frame[start..start + 4].try_into().unwrap()) + }) + .collect(); + Response { + operation_id: u64::from_le_bytes(frame[8..16].try_into().unwrap()), + path_count: u32::from_le_bytes(frame[16..20].try_into().unwrap()), + status: u32::from_le_bytes(frame[20..24].try_into().unwrap()), + failed_indices, + } + } + + /// Feeds `request` to the fsevent handler and returns the handler result plus everything the + /// handler wrote back to the peer. + async fn exchange( + request: &[u8], + dedupe: Arc, + ) -> (std::io::Result<()>, Vec) { + let (mut peer, guest) = tokio::io::duplex(256 * 1024); + peer.write_all(request).await.expect("write request"); + let result = handle_fsevent_batch(guest, dedupe).await; + let mut response = Vec::new(); + peer.read_to_end(&mut response) + .await + .expect("read response"); + (result, response) + } + #[test] fn host_only_listener_policy_accepts_host_and_rejects_guest_cids() { assert!(is_host_peer(&VsockAddr::new(VMADDR_CID_HOST, PORT_CONTROL))); assert!(!is_host_peer(&VsockAddr::new(3, PORT_CONTROL))); assert!(!is_host_peer(&VsockAddr::new(VMADDR_CID_ANY, PORT_CONTROL))); } + + #[tokio::test] + async fn a_valid_batch_is_nudged_and_answered_with_per_path_results() { + let directory = std::env::temp_dir().join(format!("dory-fsevents-{}", std::process::id())); + std::fs::create_dir_all(&directory).expect("create batch directory"); + let existing = directory.join("present"); + std::fs::write(&existing, b"payload").expect("write batch file"); + // Nudging walks up to an existing ancestor, so only a path whose every ancestor is missing + // (the filesystem root is never nudged) can fail. + let unreachable = PathBuf::from("/dory-absent-fsevent-root/child"); + let paths = vec![existing.clone(), unreachable]; + + let request = encode_batch_frame(7, &paths).expect("encode batch"); + let (result, response) = exchange(&request, Arc::default()).await; + + result.expect("handled batch"); + let decoded = decode_response(&response); + assert_eq!(decoded.operation_id, 7); + assert_eq!(decoded.path_count, 2); + assert_eq!(decoded.status, ResponseStatus::Success as u32); + // Only the path whose parent directory does not exist can fail to be nudged. + assert_eq!(decoded.failed_indices, vec![1]); + + std::fs::remove_dir_all(&directory).expect("clean batch directory"); + } + + #[tokio::test] + async fn reusing_an_operation_id_for_different_paths_is_reported_as_a_conflict() { + let dedupe = Arc::new(crate::fsevents::FSEventDedupeStore::default()); + let first = encode_batch_frame(11, &[PathBuf::from("/tmp")]).expect("encode batch"); + let second = encode_batch_frame(11, &[PathBuf::from("/var")]).expect("encode batch"); + + let (first_result, _) = exchange(&first, Arc::clone(&dedupe)).await; + first_result.expect("handled first batch"); + let (second_result, response) = exchange(&second, dedupe).await; + + second_result.expect("handled second batch"); + let decoded = decode_response(&response); + assert_eq!(decoded.operation_id, 11); + assert_eq!(decoded.path_count, 1); + assert_eq!( + decoded.status, + ResponseStatus::ConflictingOperationId as u32 + ); + assert!(decoded.failed_indices.is_empty()); + } + + #[tokio::test] + async fn oversized_and_undecodable_frames_are_rejected_without_a_response() { + let mut oversized = ((MAX_FRAME_BYTES + 1) as u32).to_le_bytes().to_vec(); + oversized.extend_from_slice(&[0_u8; 16]); + let (result, response) = exchange(&oversized, Arc::default()).await; + assert_eq!( + result.unwrap_err().kind(), + std::io::ErrorKind::InvalidData, + "declared body length above the frame cap must be refused" + ); + assert!(response.is_empty()); + + let mut wrong_version = 16_u32.to_le_bytes().to_vec(); + wrong_version.extend_from_slice(&(PROTOCOL_VERSION + 1).to_le_bytes()); + wrong_version.extend_from_slice(&1_u64.to_le_bytes()); + wrong_version.extend_from_slice(&0_u32.to_le_bytes()); + let (result, response) = exchange(&wrong_version, Arc::default()).await; + assert_eq!(result.unwrap_err().kind(), std::io::ErrorKind::InvalidData); + assert!(response.is_empty()); + } + + #[tokio::test] + async fn a_slow_drip_frame_hits_the_read_deadline() { + // The length prefix promises a body that never arrives, which is exactly the stall the + // per-connection deadline exists to bound. + let (peer, guest) = tokio::io::duplex(1024); + let mut peer = peer; + peer.write_all(&64_u32.to_le_bytes()) + .await + .expect("write length prefix"); + let result = handle_fsevent_batch(guest, Arc::default()).await; + assert_eq!(result.unwrap_err().kind(), std::io::ErrorKind::TimedOut); + drop(peer); + } } diff --git a/dory-core/transfer-helper/src/lib.rs b/dory-core/transfer-helper/src/lib.rs index 8b6cd97c..f0da78b5 100644 --- a/dory-core/transfer-helper/src/lib.rs +++ b/dory-core/transfer-helper/src/lib.rs @@ -12,6 +12,9 @@ mod scan_linux; #[cfg(target_os = "linux")] mod repair_linux; +#[cfg(all(test, target_os = "linux"))] +mod test_support; + pub use model::{ DataExtent, ManifestEntry, ManifestEntryKind, ManifestLimits, VolumeManifest, XattrEntry, }; @@ -110,3 +113,48 @@ fn hex_nibble(value: u8) -> Option { _ => None, } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn hex_round_trips_every_byte_value() { + let bytes: Vec = (0..=255_u8).collect(); + let encoded = hex_encode(&bytes); + assert_eq!(encoded.len(), 512); + assert!(encoded + .bytes() + .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte))); + assert_eq!(hex_decode(&encoded).unwrap(), bytes); + assert_eq!(hex_encode(&[]), ""); + assert_eq!(hex_decode("").unwrap(), Vec::::new()); + } + + #[test] + fn hex_decode_rejects_odd_length_and_non_canonical_digits() { + for value in ["a", "abc", "4A", "ff0G", "ff 0"] { + assert!( + matches!( + hex_decode(value), + Err(TransferHelperError::InvalidManifest(_)) + ), + "{value}" + ); + } + } + + #[test] + fn errors_report_their_path_and_source() { + let error = TransferHelperError::Filesystem { + path: "hex:2f".into(), + source: std::io::Error::from(std::io::ErrorKind::NotFound), + }; + let message = error.to_string(); + assert!(message.contains("hex:2f"), "{message}"); + assert!(std::error::Error::source(&error).is_some()); + assert!(TransferHelperError::LinuxRequired + .to_string() + .contains("Linux")); + } +} diff --git a/dory-core/transfer-helper/src/main.rs b/dory-core/transfer-helper/src/main.rs index 0d347be5..918a4b84 100644 --- a/dory-core/transfer-helper/src/main.rs +++ b/dory-core/transfer-helper/src/main.rs @@ -160,3 +160,185 @@ fn filesystem_error(path: &Path, source: std::io::Error) -> TransferHelperError source, } } + +#[cfg(test)] +mod tests { + use super::*; + use std::path::PathBuf; + use std::sync::atomic::{AtomicU32, Ordering}; + + static COUNTER: AtomicU32 = AtomicU32::new(0); + + struct TempDirectory { + path: PathBuf, + } + + impl TempDirectory { + fn new() -> Self { + let unique = COUNTER.fetch_add(1, Ordering::Relaxed); + let path = std::env::temp_dir() + .join(format!("dory-helper-cli-{}-{unique}", std::process::id())); + let _ = fs::remove_dir_all(&path); + fs::create_dir_all(&path).expect("create temporary directory"); + Self { path } + } + + fn join(&self, relative: &str) -> PathBuf { + self.path.join(relative) + } + } + + impl Drop for TempDirectory { + fn drop(&mut self) { + let _ = fs::remove_dir_all(&self.path); + } + } + + fn arguments(values: &[&str]) -> Vec { + values.iter().map(|value| (*value).to_string()).collect() + } + + fn volume(directory: &TempDirectory, name: &str) -> PathBuf { + let root = directory.join(name); + fs::create_dir_all(root.join("nested")).unwrap(); + fs::write(root.join("nested/file"), b"payload").unwrap(); + std::os::unix::fs::symlink("nested/file", root.join("link")).unwrap(); + root + } + + #[test] + fn scan_then_repair_round_trips_through_the_manifest_file() { + let directory = TempDirectory::new(); + let source = volume(&directory, "source"); + let manifest_path = directory.join("manifest.json"); + + run(arguments(&[ + "scan", + "--root", + source.to_str().unwrap(), + "--output", + manifest_path.to_str().unwrap(), + ])) + .expect("scan"); + + let limits = ManifestLimits::default(); + let manifest = read_manifest(&manifest_path, limits).expect("read manifest"); + assert_eq!(manifest.entries.len(), 3); + assert!(!manifest.contains_device_nodes()); + + let target = volume(&directory, "target"); + run(arguments(&[ + "repair", + "--root", + target.to_str().unwrap(), + "--manifest", + manifest_path.to_str().unwrap(), + ])) + .expect("repair"); + + assert_eq!( + scan_volume(&target, limits) + .unwrap() + .canonical_sha256(limits) + .unwrap(), + manifest.canonical_sha256(limits).unwrap() + ); + } + + #[test] + fn unrecognized_argument_shapes_report_the_usage_contract() { + let rejected: Vec> = vec![ + arguments(&[]), + arguments(&["scan"]), + arguments(&["scan", "--root", "/tmp", "--out", "/tmp/manifest"]), + arguments(&["repair", "--root", "/tmp", "--output", "/tmp/manifest"]), + arguments(&["verify", "--root", "/tmp", "--manifest", "/tmp/manifest"]), + arguments(&[ + "scan", + "--root", + "/tmp", + "--output", + "/tmp/manifest", + "extra", + ]), + ]; + for value in rejected { + match run(value.clone()) { + Err(TransferHelperError::InvalidManifest(message)) => { + assert!(message.starts_with("usage:"), "{message}") + } + other => panic!("{value:?} was not rejected: {other:?}"), + } + } + } + + #[test] + fn manifest_input_must_be_a_canonical_singly_linked_regular_file() { + let directory = TempDirectory::new(); + let limits = ManifestLimits::default(); + + let missing = directory.join("absent.json"); + assert!(matches!( + read_manifest(&missing, limits), + Err(TransferHelperError::Filesystem { .. }) + )); + + let not_json = directory.join("not-json.json"); + fs::write(¬_json, b"{").unwrap(); + assert!(matches!( + read_manifest(¬_json, limits), + Err(TransferHelperError::Encoding(_)) + )); + + let source = volume(&directory, "source"); + let canonical = directory.join("manifest.json"); + let manifest = scan_volume(&source, limits).unwrap(); + write_atomic(&canonical, &manifest.canonical_json(limits).unwrap()).unwrap(); + assert_eq!(read_manifest(&canonical, limits).unwrap(), manifest); + + let reformatted = directory.join("reformatted.json"); + fs::write(&reformatted, serde_json::to_vec_pretty(&manifest).unwrap()).unwrap(); + assert!(matches!( + read_manifest(&reformatted, limits), + Err(TransferHelperError::InvalidManifest(_)) + )); + + let oversized = ManifestLimits { + maximum_manifest_bytes: 1, + ..limits + }; + assert!(matches!( + read_manifest(&canonical, oversized), + Err(TransferHelperError::Limit(_)) + )); + + let linked = directory.join("linked.json"); + fs::hard_link(&canonical, &linked).unwrap(); + assert!(matches!( + read_manifest(&linked, limits), + Err(TransferHelperError::InvalidManifest(_)) + )); + } + + #[test] + fn atomic_writes_replace_stale_partials_and_reject_unusable_paths() { + let directory = TempDirectory::new(); + let output = directory.join("receipt.json"); + fs::write(directory.join(".receipt.json.partial"), b"stale").unwrap(); + + write_atomic(&output, b"first").unwrap(); + assert_eq!(fs::read(&output).unwrap(), b"first"); + write_atomic(&output, b"second").unwrap(); + assert_eq!(fs::read(&output).unwrap(), b"second"); + assert!(!directory.join(".receipt.json.partial").exists()); + + assert!(matches!( + write_atomic(Path::new("/"), b"payload"), + Err(TransferHelperError::InvalidManifest(_)) + )); + assert!(matches!( + write_atomic(&directory.join("absent/receipt.json"), b"payload"), + Err(TransferHelperError::Filesystem { .. }) + )); + } +} diff --git a/dory-core/transfer-helper/src/repair_linux.rs b/dory-core/transfer-helper/src/repair_linux.rs index 94e70e29..3f5943aa 100644 --- a/dory-core/transfer-helper/src/repair_linux.rs +++ b/dory-core/transfer-helper/src/repair_linux.rs @@ -434,3 +434,288 @@ fn filesystem_error(path: &Path, source: std::io::Error) -> TransferHelperError source, } } + +#[cfg(test)] +mod tests { + use super::*; + use crate::test_support::{make_fifo, permissions, try_set_xattr, write_file, TempRoot}; + use crate::{hex_encode, DataExtent}; + use sha2::{Digest, Sha256}; + use std::fs; + + fn limits() -> ManifestLimits { + ManifestLimits::default() + } + + /// The archive transport reproduces names and bytes; modes, ownership, xattrs, and timestamps + /// are what `repair_volume` has to converge. `decorate` therefore only varies metadata. + fn build_volume(volume: &TempRoot, decorate: bool) { + fs::create_dir(volume.join("dir")).unwrap(); + write_file(&volume.join("dir/file"), b"payload"); + fs::hard_link(volume.join("dir/file"), volume.join("dir/link")).unwrap(); + std::os::unix::fs::symlink("dir/file", volume.join("symlink")).unwrap(); + make_fifo(&volume.join("pipe")); + if decorate { + fs::set_permissions(volume.join("dir"), permissions(0o750)).unwrap(); + fs::set_permissions(volume.join("dir/file"), permissions(0o640)).unwrap(); + fs::set_permissions(volume.join("pipe"), permissions(0o600)).unwrap(); + let _ = try_set_xattr(&volume.join("dir/file"), "user.source", b"kept"); + } else { + fs::set_permissions(volume.join("dir"), permissions(0o700)).unwrap(); + fs::set_permissions(volume.join("dir/file"), permissions(0o600)).unwrap(); + let _ = try_set_xattr(&volume.join("dir/file"), "user.stale", b"removed"); + } + } + + fn entry(path: &str, kind: ManifestEntryKind) -> ManifestEntry { + ManifestEntry { + path_hex: hex_encode(path.as_bytes()), + kind, + mode: 0o644, + uid: unsafe { libc::getuid() }, + gid: unsafe { libc::getgid() }, + size: 0, + mtime_seconds: 1, + mtime_nanoseconds: 2, + content_sha256: None, + link_target_hex: None, + hard_link_target_hex: None, + sparse_data_extents: None, + device_major: None, + device_minor: None, + xattrs: vec![], + } + } + + fn root_entry() -> ManifestEntry { + let mut value = entry("", ManifestEntryKind::Directory); + value.mode = 0o755; + value + } + + #[test] + fn repair_converges_the_target_onto_the_source_manifest() { + let source = TempRoot::new(); + build_volume(&source, true); + let expected = crate::scan_volume(source.path(), limits()).unwrap(); + + let target = TempRoot::new(); + build_volume(&target, false); + fs::set_permissions(target.path(), permissions(0o700)).unwrap(); + + let repaired = repair_volume(target.path(), &expected, limits()).unwrap(); + assert_eq!(repaired, expected.normalized_target()); + assert_eq!( + crate::scan_volume(target.path(), limits()).unwrap(), + expected + ); + assert_eq!( + repaired.canonical_sha256(limits()).unwrap(), + expected.canonical_sha256(limits()).unwrap() + ); + } + + #[test] + fn repair_restores_sparse_holes_for_regular_files() { + let target = TempRoot::new(); + let path = target.join("sparse"); + write_file(&path, &vec![0_u8; 512 * 1024]); + + let mut file = entry("sparse", ManifestEntryKind::RegularFile); + file.size = 512 * 1024; + file.content_sha256 = Some(hex_encode(&Sha256::digest(vec![0_u8; 512 * 1024]))); + file.sparse_data_extents = Some(vec![]); + let manifest = VolumeManifest::new(root_entry(), vec![file]); + + let repaired = repair_volume(target.path(), &manifest, limits()).unwrap(); + let entry = &repaired.entries[0]; + assert_eq!(entry.size, 512 * 1024); + assert_eq!(entry.sparse_data_extents.as_deref(), Some(&[][..])); + assert_eq!(fs::read(&path).unwrap(), vec![0_u8; 512 * 1024]); + } + + #[test] + fn sockets_in_the_source_are_excluded_from_the_repaired_target() { + let source = TempRoot::new(); + write_file(&source.join("file"), b"payload"); + let listener = std::os::unix::net::UnixListener::bind(source.join("service.sock")).unwrap(); + let manifest = crate::scan_volume(source.path(), limits()).unwrap(); + drop(listener); + assert_eq!(manifest.socket_paths().len(), 1); + + let target = TempRoot::new(); + write_file(&target.join("file"), b"payload"); + + let repaired = repair_volume(target.path(), &manifest, limits()).unwrap(); + assert_eq!(repaired.entries.len(), 1); + assert!(repaired.socket_paths().is_empty()); + } + + #[test] + fn device_nodes_are_refused_by_the_launch_policy() { + let target = TempRoot::new(); + let mut device = entry("device", ManifestEntryKind::CharacterDevice); + device.device_major = Some(1); + device.device_minor = Some(3); + let manifest = VolumeManifest::new(root_entry(), vec![device]); + + assert!(matches!( + repair_volume(target.path(), &manifest, limits()), + Err(TransferHelperError::Unsupported(_)) + )); + } + + #[test] + fn an_invalid_source_manifest_never_touches_the_target() { + let target = TempRoot::new(); + let mut broken = root_entry(); + broken.mtime_nanoseconds = 1_000_000_000; + let manifest = VolumeManifest::new(broken, vec![]); + assert!(matches!( + repair_volume(target.path(), &manifest, limits()), + Err(TransferHelperError::InvalidManifest(_)) + )); + } + + #[test] + fn transport_verification_rejects_missing_extra_and_altered_paths() { + let target = TempRoot::new(); + write_file(&target.join("file"), b"payload"); + + let mut file = entry("file", ManifestEntryKind::RegularFile); + file.size = 7; + file.content_sha256 = Some(hex_encode(&Sha256::digest(b"payload"))); + file.sparse_data_extents = Some(vec![DataExtent { + offset: 0, + length: 7, + }]); + + let mut extra = entry("absent", ManifestEntryKind::Directory); + extra.mode = 0o755; + let too_many = VolumeManifest::new(root_entry(), vec![file.clone(), extra]); + assert!(matches!( + repair_volume(target.path(), &too_many, limits()), + Err(TransferHelperError::Verification(_)) + )); + + let mut renamed = file.clone(); + renamed.path_hex = hex_encode(b"other"); + let wrong_path = VolumeManifest::new(root_entry(), vec![renamed]); + assert!(matches!( + repair_volume(target.path(), &wrong_path, limits()), + Err(TransferHelperError::Verification(_)) + )); + + let mut wrong_kind = file.clone(); + wrong_kind.kind = ManifestEntryKind::Fifo; + wrong_kind.size = 0; + wrong_kind.content_sha256 = None; + wrong_kind.sparse_data_extents = None; + let wrong_kind = VolumeManifest::new(root_entry(), vec![wrong_kind]); + assert!(matches!( + repair_volume(target.path(), &wrong_kind, limits()), + Err(TransferHelperError::Verification(_)) + )); + + let mut wrong_content = file; + wrong_content.content_sha256 = Some(hex_encode(&Sha256::digest(b"different"))); + let wrong_content = VolumeManifest::new(root_entry(), vec![wrong_content]); + assert!(matches!( + repair_volume(target.path(), &wrong_content, limits()), + Err(TransferHelperError::Verification(_)) + )); + } + + #[test] + fn symbolic_and_hard_link_targets_are_verified_against_the_target() { + let target = TempRoot::new(); + write_file(&target.join("file"), b"payload"); + std::os::unix::fs::symlink("file", target.join("symlink")).unwrap(); + fs::hard_link(target.join("file"), target.join("linked")).unwrap(); + + let observed = crate::scan_volume(target.path(), limits()).unwrap(); + let mut drifted = observed.clone(); + for entry in &mut drifted.entries { + match entry.kind { + ManifestEntryKind::SymbolicLink => { + entry.link_target_hex = Some(hex_encode(b"other")); + entry.size = 5; + } + ManifestEntryKind::HardLink => { + entry.hard_link_target_hex = Some(hex_encode(b"absent")); + } + _ => {} + } + } + assert!(matches!( + verify_transport_content(&drifted, &observed), + Err(TransferHelperError::Verification(_)) + )); + verify_transport_content(&observed, &observed).unwrap(); + } + + #[test] + fn manifest_paths_decode_byte_components_under_the_root() { + let root = Path::new("/volume"); + assert_eq!( + manifest_path(root, &root_entry()).unwrap(), + Path::new("/volume") + ); + let nested = entry("dir/file", ManifestEntryKind::Directory); + assert_eq!( + manifest_path(root, &nested).unwrap(), + Path::new("/volume/dir/file") + ); + let mut invalid = nested; + invalid.path_hex = "abc".into(); + assert!(matches!( + manifest_path(root, &invalid), + Err(TransferHelperError::InvalidManifest(_)) + )); + } + + #[test] + fn exact_mismatch_reports_the_first_differing_entry() { + let file = { + let mut value = entry("file", ManifestEntryKind::RegularFile); + value.size = 7; + value.content_sha256 = Some(hex_encode(&Sha256::digest(b"payload"))); + value.sparse_data_extents = Some(vec![DataExtent { + offset: 0, + length: 7, + }]); + value + }; + let expected = VolumeManifest::new(root_entry(), vec![file.clone()]); + + let mut different_root = expected.clone(); + different_root.root.mode = 0o700; + assert_eq!( + first_exact_mismatch(&expected, &different_root), + "target root metadata does not match the source manifest" + ); + + let mut different_entry = expected.clone(); + different_entry.entries[0].mode = 0o600; + assert_eq!( + first_exact_mismatch(&expected, &different_entry), + format!("target differs at hex:{}", file.path_hex) + ); + + let empty = VolumeManifest::new(root_entry(), vec![]); + assert_eq!( + first_exact_mismatch(&expected, &empty), + "target entry count differs: expected 1, found 0" + ); + } + + #[test] + fn punching_a_zero_length_hole_is_a_no_op() { + let target = TempRoot::new(); + let path = target.join("file"); + write_file(&path, b"payload"); + let file = fs::OpenOptions::new().write(true).open(&path).unwrap(); + punch_hole(&file, 0, 0, &path).unwrap(); + assert_eq!(fs::read(&path).unwrap(), b"payload"); + } +} diff --git a/dory-core/transfer-helper/src/scan_linux.rs b/dory-core/transfer-helper/src/scan_linux.rs index 76a628f3..19c70e4e 100644 --- a/dory-core/transfer-helper/src/scan_linux.rs +++ b/dory-core/transfer-helper/src/scan_linux.rs @@ -460,3 +460,284 @@ fn path_display_lossless(path: &Path) -> String { fn _os_name_bytes(value: &OsStr) -> &[u8] { value.as_bytes() } + +#[cfg(test)] +mod tests { + use super::*; + use crate::test_support::{make_fifo, try_set_xattr, write_file, write_sparse_file, TempRoot}; + use std::os::unix::ffi::OsStringExt; + + fn scan(root: &Path) -> VolumeManifest { + scan_volume(root, ManifestLimits::default()).expect("scan volume") + } + + fn entry<'a>(manifest: &'a VolumeManifest, path: &[u8]) -> &'a ManifestEntry { + let path_hex = hex_encode(path); + manifest + .entries + .iter() + .find(|entry| entry.path_hex == path_hex) + .unwrap_or_else(|| panic!("missing entry hex:{path_hex}")) + } + + #[test] + fn scan_describes_every_supported_file_type_and_validates() { + let volume = TempRoot::new(); + fs::create_dir(volume.join("dir")).unwrap(); + write_file(&volume.join("dir/file"), b"payload"); + std::os::unix::fs::symlink("dir/file", volume.join("link")).unwrap(); + make_fifo(&volume.join("pipe")); + let binary_name = std::ffi::OsString::from_vec(b"non\xffutf8".to_vec()); + write_file(&volume.path().join(binary_name), b"raw"); + + let manifest = scan(volume.path()); + manifest.validate(ManifestLimits::default()).unwrap(); + + assert_eq!(manifest.root.path_hex, ""); + assert_eq!(manifest.root.kind, ManifestEntryKind::Directory); + assert_eq!(manifest.entries.len(), 5); + assert!(manifest + .entries + .windows(2) + .all(|pair| pair[0].path_hex < pair[1].path_hex)); + + let directory = entry(&manifest, b"dir"); + assert_eq!(directory.kind, ManifestEntryKind::Directory); + assert_eq!(directory.size, 0); + assert!(directory.content_sha256.is_none()); + + let file = entry(&manifest, b"dir/file"); + assert_eq!(file.kind, ManifestEntryKind::RegularFile); + assert_eq!(file.size, 7); + assert_eq!( + file.content_sha256.as_deref(), + Some(hex_encode(&Sha256::digest(b"payload")).as_str()) + ); + assert_eq!( + file.sparse_data_extents.as_deref(), + Some( + &[DataExtent { + offset: 0, + length: 7 + }][..] + ) + ); + + let link = entry(&manifest, b"link"); + assert_eq!(link.kind, ManifestEntryKind::SymbolicLink); + assert_eq!( + link.link_target_hex.as_deref(), + Some(hex_encode(b"dir/file").as_str()) + ); + assert_eq!(link.size, "dir/file".len() as u64); + assert!(link.content_sha256.is_none()); + + assert_eq!(entry(&manifest, b"pipe").kind, ManifestEntryKind::Fifo); + assert_eq!( + entry(&manifest, b"non\xffutf8").kind, + ManifestEntryKind::RegularFile + ); + } + + #[test] + fn multiply_linked_regular_files_collapse_into_one_canonical_inode() { + let volume = TempRoot::new(); + write_file(&volume.join("aaa-original"), b"same inode"); + fs::hard_link(volume.join("aaa-original"), volume.join("zzz-link")).unwrap(); + + let manifest = scan(volume.path()); + manifest.validate(ManifestLimits::default()).unwrap(); + + let original = entry(&manifest, b"aaa-original"); + assert_eq!(original.kind, ManifestEntryKind::RegularFile); + let link = entry(&manifest, b"zzz-link"); + assert_eq!(link.kind, ManifestEntryKind::HardLink); + assert_eq!( + link.hard_link_target_hex.as_deref(), + Some(original.path_hex.as_str()) + ); + assert!(link.content_sha256.is_none()); + assert!(link.sparse_data_extents.is_none()); + } + + #[test] + fn sockets_are_scanned_so_the_target_contract_can_exclude_them() { + let volume = TempRoot::new(); + let _listener = + std::os::unix::net::UnixListener::bind(volume.join("service.sock")).unwrap(); + + let manifest = scan(volume.path()); + manifest.validate(ManifestLimits::default()).unwrap(); + assert_eq!( + entry(&manifest, b"service.sock").kind, + ManifestEntryKind::Socket + ); + assert_eq!(manifest.socket_paths(), vec![hex_encode(b"service.sock")]); + assert!(manifest.normalized_target().entries.is_empty()); + } + + #[test] + fn sparse_regions_are_proven_as_data_extents() { + let volume = TempRoot::new(); + let path = volume.join("sparse"); + write_sparse_file(&path, 1024 * 1024, b"tail"); + + let manifest = scan(volume.path()); + manifest.validate(ManifestLimits::default()).unwrap(); + let file = entry(&manifest, b"sparse"); + assert_eq!(file.size, 1024 * 1024 + 4); + let extents = file.sparse_data_extents.as_deref().unwrap(); + assert!(!extents.is_empty()); + assert!(extents + .iter() + .all(|extent| extent.offset + extent.length <= file.size)); + assert_eq!( + extents.last().map(|extent| extent.offset + extent.length), + Some(file.size) + ); + + write_file(&volume.join("empty"), b""); + let manifest = scan(volume.path()); + let empty = entry(&manifest, b"empty"); + assert_eq!(empty.size, 0); + assert_eq!(empty.sparse_data_extents.as_deref(), Some(&[][..])); + } + + #[test] + fn xattrs_are_recorded_in_sorted_hex_form() { + let volume = TempRoot::new(); + let path = volume.join("file"); + write_file(&path, b"payload"); + if !try_set_xattr(&path, "user.zeta", b"\x00\xff") + || !try_set_xattr(&path, "user.alpha", b"first") + { + return; + } + + let manifest = scan(volume.path()); + manifest.validate(ManifestLimits::default()).unwrap(); + let file = entry(&manifest, b"file"); + assert_eq!( + file.xattrs, + vec![ + XattrEntry { + name_hex: hex_encode(b"user.alpha"), + value_hex: hex_encode(b"first"), + }, + XattrEntry { + name_hex: hex_encode(b"user.zeta"), + value_hex: hex_encode(b"\x00\xff"), + }, + ] + ); + + let limits = ManifestLimits { + maximum_xattrs_per_entry: 1, + ..ManifestLimits::default() + }; + assert!(matches!( + scan_volume(volume.path(), limits), + Err(TransferHelperError::Limit(_)) + )); + + let limits = ManifestLimits { + maximum_xattr_value_bytes: 1, + ..ManifestLimits::default() + }; + assert!(matches!( + scan_volume(volume.path(), limits), + Err(TransferHelperError::Limit(_)) + )); + } + + #[test] + fn scan_fails_closed_on_a_non_directory_root_and_a_missing_root() { + let volume = TempRoot::new(); + let file = volume.join("file"); + write_file(&file, b"payload"); + assert!(matches!( + scan_volume(&file, ManifestLimits::default()), + Err(TransferHelperError::InvalidManifest(_)) + )); + assert!(matches!( + scan_volume(&volume.join("absent"), ManifestLimits::default()), + Err(TransferHelperError::Filesystem { .. }) + )); + } + + #[test] + fn entry_and_path_limits_stop_the_walk() { + let volume = TempRoot::new(); + fs::create_dir(volume.join("nested")).unwrap(); + write_file(&volume.join("nested/file"), b"payload"); + write_file(&volume.join("other"), b"payload"); + + let limits = ManifestLimits { + maximum_entries: 2, + ..ManifestLimits::default() + }; + assert!(matches!( + scan_volume(volume.path(), limits), + Err(TransferHelperError::Limit(_)) + )); + + let limits = ManifestLimits { + maximum_path_bytes: 6, + ..ManifestLimits::default() + }; + assert!(matches!( + scan_volume(volume.path(), limits), + Err(TransferHelperError::Limit(_)) + )); + } + + #[test] + fn relative_paths_join_with_byte_semantics_and_reject_oversized_results() { + assert_eq!(join_relative(b"", b"child", 4096).unwrap(), b"child"); + assert_eq!( + join_relative(b"parent", b"child", 4096).unwrap(), + b"parent/child" + ); + assert_eq!(join_relative(b"a", b"b", 3).unwrap(), b"a/b"); + assert!(matches!( + join_relative(b"a", b"b", 2), + Err(TransferHelperError::Limit(_)) + )); + assert!(join_relative(b"a", b"b", usize::MAX).is_ok()); + } + + #[test] + fn drift_between_the_expected_and_observed_inode_is_reported() { + let volume = TempRoot::new(); + let path = volume.join("file"); + write_file(&path, b"payload"); + let metadata = fs::symlink_metadata(&path).unwrap(); + write_file(&path, b"a longer payload"); + assert!(matches!( + hash_and_extents(&path, &metadata, b"file"), + Err(TransferHelperError::SourceDrift(_)) + )); + } + + #[test] + fn paths_are_reported_losslessly_as_hex() { + assert_eq!(display_path(b""), "."); + assert_eq!(display_path(b"a/b"), format!("hex:{}", hex_encode(b"a/b"))); + assert!(matches!( + path_c_string(Path::new(OsStr::from_bytes(b"nul\0path"))), + Err(TransferHelperError::InvalidManifest(_)) + )); + assert_eq!( + path_display_lossless(Path::new("ab")), + format!("hex:{}", hex_encode(b"ab")) + ); + } + + #[test] + fn unsupported_sparse_errors_become_an_unsupported_kind() { + let translated = sparse_error(std::io::Error::from_raw_os_error(libc::EINVAL)); + assert_eq!(translated.kind(), std::io::ErrorKind::Unsupported); + let preserved = sparse_error(std::io::Error::from_raw_os_error(libc::EIO)); + assert_eq!(preserved.raw_os_error(), Some(libc::EIO)); + } +} diff --git a/dory-core/transfer-helper/src/test_support.rs b/dory-core/transfer-helper/src/test_support.rs new file mode 100644 index 00000000..4b258e16 --- /dev/null +++ b/dory-core/transfer-helper/src/test_support.rs @@ -0,0 +1,112 @@ +//! Shared scaffolding for the Linux scan and repair tests: a self-cleaning temporary volume plus +//! the raw filesystem primitives the manifest contract has to describe. + +use std::ffi::CString; +use std::fs; +use std::io::{Seek, SeekFrom, Write}; +use std::os::unix::ffi::OsStrExt; +use std::path::{Path, PathBuf}; +use std::sync::atomic::{AtomicU32, Ordering}; + +static COUNTER: AtomicU32 = AtomicU32::new(0); + +/// Temporary volumes live in `TMPDIR` when that filesystem can hold user xattrs, and under the +/// build directory otherwise: tmpfs rejects `user.*` xattrs on older kernels, and the manifest +/// contract covers xattrs. +fn base_directory() -> PathBuf { + let candidate = std::env::temp_dir(); + #[allow(clippy::disallowed_names)] + let probe = candidate.join(format!("dory-xattr-probe-{}", std::process::id())); + let supported = fs::File::create(&probe) + .map(|_| try_set_xattr(&probe, "user.dory-probe", b"1")) + .unwrap_or(false); + let _ = fs::remove_file(&probe); + if supported { + candidate + } else { + Path::new(env!("CARGO_MANIFEST_DIR")).join("../target/dory-transfer-tests") + } +} + +pub struct TempRoot { + path: PathBuf, +} + +impl Default for TempRoot { + fn default() -> Self { + Self::new() + } +} + +impl TempRoot { + /// The name stays short because tests bind Unix sockets inside the volume, and `sockaddr_un` + /// paths are limited to 108 bytes. + pub fn new() -> Self { + let unique = COUNTER.fetch_add(1, Ordering::Relaxed); + let base = base_directory(); + fs::create_dir_all(&base).expect("create temporary volume base"); + let path = base.join(format!("v{}-{unique}", std::process::id())); + let _ = fs::remove_dir_all(&path); + fs::create_dir_all(&path).expect("create temporary volume"); + fs::set_permissions(&path, permissions(0o755)).expect("set root mode"); + Self { path } + } + + pub fn path(&self) -> &Path { + &self.path + } + + pub fn join(&self, relative: &str) -> PathBuf { + self.path.join(relative) + } +} + +impl Drop for TempRoot { + fn drop(&mut self) { + let _ = fs::remove_dir_all(&self.path); + } +} + +pub fn permissions(mode: u32) -> fs::Permissions { + use std::os::unix::fs::PermissionsExt; + fs::Permissions::from_mode(mode) +} + +pub fn write_file(path: &Path, contents: &[u8]) { + fs::write(path, contents).expect("write file"); +} + +pub fn write_sparse_file(path: &Path, hole_bytes: u64, tail: &[u8]) { + let mut file = fs::File::create(path).expect("create sparse file"); + file.seek(SeekFrom::Start(hole_bytes)).expect("seek hole"); + file.write_all(tail).expect("write sparse tail"); + file.sync_all().expect("sync sparse file"); +} + +pub fn make_fifo(path: &Path) { + let raw = CString::new(path.as_os_str().as_bytes()).expect("fifo path has no nul"); + let result = unsafe { libc::mkfifo(raw.as_ptr(), 0o644) }; + assert_eq!( + result, + 0, + "mkfifo failed: {}", + std::io::Error::last_os_error() + ); +} + +/// Returns false when the filesystem backing the temporary volume rejects user xattrs, which +/// happens on older tmpfs kernels. Callers skip their xattr assertions in that case. +pub fn try_set_xattr(path: &Path, name: &str, value: &[u8]) -> bool { + let raw_path = CString::new(path.as_os_str().as_bytes()).expect("xattr path has no nul"); + let raw_name = CString::new(name).expect("xattr name has no nul"); + let result = unsafe { + libc::lsetxattr( + raw_path.as_ptr(), + raw_name.as_ptr(), + value.as_ptr().cast::(), + value.len(), + 0, + ) + }; + result == 0 +}