diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index 20eb7fd607..44ddf7b894 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -4556,6 +4556,7 @@ dependencies = [ "codex-utils-path", "codex-utils-process", "codex-utils-string", + "libc", "memchr", "pretty_assertions", "regex", diff --git a/codex-rs/rollout/Cargo.toml b/codex-rs/rollout/Cargo.toml index 6bb676c9b6..2bd21c9251 100644 --- a/codex-rs/rollout/Cargo.toml +++ b/codex-rs/rollout/Cargo.toml @@ -51,6 +51,9 @@ tracing = { workspace = true } uuid = { workspace = true } zstd = { workspace = true } +[target.'cfg(unix)'.dependencies] +libc = { workspace = true } + [dev-dependencies] codex-utils-absolute-path = { workspace = true } pretty_assertions = { workspace = true } diff --git a/codex-rs/rollout/src/compression.rs b/codex-rs/rollout/src/compression.rs index dc4ffe1df5..c0f33f14af 100644 --- a/codex-rs/rollout/src/compression.rs +++ b/codex-rs/rollout/src/compression.rs @@ -1,21 +1,16 @@ -use std::ffi::OsStr; use std::fs::File; -use std::fs::Permissions; use std::io; use std::io::Read; use std::io::Write; use std::path::Path; use std::path::PathBuf; -use std::sync::atomic::AtomicU64; -use std::sync::atomic::Ordering; use std::time::Duration; use std::time::Instant; #[cfg(unix)] use std::os::unix::fs::OpenOptionsExt; -#[cfg(unix)] -use std::os::unix::fs::PermissionsExt; +mod access_metadata; mod blocking_reader; mod error_metrics; mod path_metadata; @@ -30,8 +25,6 @@ use read_metrics::ReadMetrics; const COMPRESSED_SUFFIX: &str = ".zst"; const MAX_NOT_FOUND_RETRIES: usize = 3; const OPEN_ROLLOUT_LINE_READER_RETRY_DELAY: Duration = Duration::from_millis(50); -const TEMP_SUFFIX: &str = ".tmp"; -static TEMP_COUNTER: AtomicU64 = AtomicU64::new(0); /// The entry point that requested a compression pass, not a Statsig cohort. #[derive(Clone, Copy, Debug)] @@ -131,40 +124,45 @@ pub(crate) fn materialize_rollout_for_append_blocking(path: &Path) -> io::Result } let started_at = Instant::now(); - let temp_path = temp_path_for(plain_path.as_path(), "decompress"); let mut stage = "prepare_directory"; let result: io::Result<()> = (|| { if let Some(parent) = plain_path.parent() { std::fs::create_dir_all(parent)?; } - stage = "read_metadata"; - let metadata = std::fs::metadata(compressed_path.as_path())?; - let permissions = metadata.permissions(); + stage = "open_source"; + let input = File::open(compressed_path.as_path())?; stage = "create_temp"; - let mut output = create_file_with_permissions(temp_path.as_path(), &permissions)?; + // Restoring mtime makes this file look stale before publication. Older + // workers delete *.tmp without coordinating with materialization. + let mut output = tempfile::Builder::new() + .prefix(".rollout-materialize-") + .make_in( + plain_path.parent().unwrap_or_else(|| Path::new(".")), + |path| { + // Hard-link publication must not retain Windows' temporary-file attribute. + let mut options = std::fs::OpenOptions::new(); + options.write(true).create_new(true); + #[cfg(unix)] + options.mode(0o600); + options.open(path) + }, + )?; { - stage = "open_source"; - let input = File::open(compressed_path.as_path())?; stage = "decode_and_write"; - let mut decoder = zstd::stream::read::Decoder::new(input)?; + let mut decoder = zstd::stream::read::Decoder::new(&input)?; io::copy(&mut decoder, &mut output)?; } stage = "flush"; output.flush()?; - stage = "sync"; - output.sync_all()?; - stage = "publish"; - match std::fs::hard_link(temp_path.as_path(), plain_path.as_path()) { - Ok(()) => {} - Err(err) if err.kind() == io::ErrorKind::AlreadyExists => {} - Err(_) => persist_temp_file_noclobber(temp_path.as_path(), plain_path.as_path())?, - } stage = "set_metadata"; - output.set_times(std::fs::FileTimes::new().set_modified(metadata.modified()?))?; + access_metadata::copy(&input, output.as_file())?; + stage = "verify"; + File::open(output.path())?; stage = "sync"; - output.sync_all()?; - drop(output); - let _ = std::fs::remove_file(temp_path.as_path()); + output.as_file().sync_all()?; + stage = "publish"; + persist_temp_file_noclobber(output, plain_path.as_path())?; + drop(input); stage = "remove_source"; match std::fs::remove_file(compressed_path.as_path()) { Ok(()) => {} @@ -174,7 +172,6 @@ pub(crate) fn materialize_rollout_for_append_blocking(path: &Path) -> io::Result Ok(()) })(); if let Err(err) = &result { - let _ = std::fs::remove_file(temp_path.as_path()); FailureMetric::Materialize.record(stage, err); metrics::materialize_duration("failed", started_at.elapsed()); } @@ -184,10 +181,17 @@ pub(crate) fn materialize_rollout_for_append_blocking(path: &Path) -> io::Result Ok(plain_path) } -fn persist_temp_file_noclobber(temp_path: &Path, destination: &Path) -> io::Result<()> { - let temp_path = tempfile::TempPath::try_from_path(temp_path)?; - match temp_path.persist_noclobber(destination) { - Ok(()) => Ok(()), +fn persist_temp_file_noclobber( + temp_file: tempfile::NamedTempFile, + destination: &Path, +) -> io::Result<()> { + match std::fs::hard_link(temp_file.path(), destination) { + Ok(()) => return Ok(()), + Err(err) if err.kind() == io::ErrorKind::AlreadyExists => return Ok(()), + Err(_) => {} + } + match temp_file.persist_noclobber(destination) { + Ok(_) => Ok(()), Err(err) if err.error.kind() == io::ErrorKind::AlreadyExists => Ok(()), Err(err) => Err(err.error), } @@ -333,10 +337,11 @@ type BlockingLineReader = std::io::Lines mod worker { use std::ffi::OsStr; use std::fs::File; - use std::fs::FileTimes; use std::fs::Permissions; use std::io; use std::io::Write; + #[cfg(unix)] + use std::os::unix::fs::MetadataExt; use std::path::Path; use std::path::PathBuf; use std::sync::Arc; @@ -355,6 +360,7 @@ mod worker { use super::RolloutCompressionTrigger; use super::RolloutFile; + use super::access_metadata; use super::error_metrics::FailureMetric; use super::metrics; use super::path; @@ -882,14 +888,14 @@ mod worker { .suffix(TEMP_SUFFIX) .tempfile_in(temp_dir) .inspect_err(|err| FailureMetric::File(trigger).record("create_temp", err))?; - encode_zstd_to_writer(path, temp_file.as_file_mut()) + let input = File::open(path) + .inspect_err(|err| FailureMetric::File(trigger).record("open_source", err))?; + encode_zstd_to_writer(&input, temp_file.as_file_mut()) .inspect_err(|err| FailureMetric::File(trigger).record("encode_and_write", err))?; temp_file .as_file_mut() .flush() .inspect_err(|err| FailureMetric::File(trigger).record("flush", err))?; - verify_zstd(temp_file.path()) - .inspect_err(|err| FailureMetric::File(trigger).record("verify", err))?; if !same_file_state(path, &before) .inspect_err(|err| FailureMetric::File(trigger).record("recheck_source", err))? { @@ -899,8 +905,12 @@ mod worker { /*compressed_bytes*/ None, )); } - set_file_metadata(temp_file.as_file(), before.modified, &before.permissions) + access_metadata::copy(&input, temp_file.as_file()) .inspect_err(|err| FailureMetric::File(trigger).record("set_metadata", err))?; + // Reopen after applying final access metadata: verifying the initial + // owner-only temporary would miss an inaccessible published file. + verify_zstd(temp_file.path()) + .inspect_err(|err| FailureMetric::File(trigger).record("verify", err))?; temp_file .as_file() .sync_all() @@ -970,6 +980,8 @@ mod worker { len: u64, modified: SystemTime, permissions: Permissions, + #[cfg(unix)] + change_time: (i64, i64), } fn cold_file_state(path: &Path) -> io::Result { @@ -988,6 +1000,8 @@ mod worker { len: metadata.len(), modified, permissions: metadata.permissions(), + #[cfg(unix)] + change_time: (metadata.ctime(), metadata.ctime_nsec()), }; let age = SystemTime::now() .duration_since(modified) @@ -1000,16 +1014,23 @@ mod worker { fn same_file_state(path: &Path, expected: &FileState) -> io::Result { match std::fs::metadata(path) { - Ok(metadata) => Ok(metadata.len() == expected.len - && metadata.modified()? == expected.modified - && metadata.permissions() == expected.permissions), + Ok(metadata) => { + // Ownership and ACL edits change ctime even when mode, length + // and mtime are unchanged. Do not publish stale access metadata. + #[cfg(unix)] + if (metadata.ctime(), metadata.ctime_nsec()) != expected.change_time { + return Ok(false); + } + Ok(metadata.len() == expected.len + && metadata.modified()? == expected.modified + && metadata.permissions() == expected.permissions) + } Err(err) if err.kind() == io::ErrorKind::NotFound => Ok(false), Err(err) => Err(err), } } - fn encode_zstd_to_writer(source: &Path, output: impl Write) -> io::Result<()> { - let mut input = File::open(source)?; + fn encode_zstd_to_writer(mut input: &File, output: impl Write) -> io::Result<()> { let mut encoder = zstd::stream::write::Encoder::new(output, COMPRESSION_LEVEL)?; // Preserve fast byte-bound checks for paginated history without decoding the whole file. encoder.set_pledged_src_size(Some(input.metadata()?.len()))?; @@ -1026,15 +1047,6 @@ mod worker { Ok(()) } - fn set_file_metadata( - file: &File, - modified: SystemTime, - permissions: &Permissions, - ) -> io::Result<()> { - file.set_times(FileTimes::new().set_modified(modified))?; - file.set_permissions(permissions.clone()) - } - async fn cleanup_stale_temps( codex_home: &Path, trigger: RolloutCompressionTrigger, @@ -1395,40 +1407,6 @@ mod reader { } } -#[cfg(unix)] -fn create_file_with_permissions(path: &Path, permissions: &Permissions) -> io::Result { - let file = std::fs::OpenOptions::new() - .write(true) - .create_new(true) - .mode(permissions.mode() & 0o7777) - .open(path)?; - file.set_permissions(permissions.clone())?; - Ok(file) -} - -#[cfg(not(unix))] -fn create_file_with_permissions(path: &Path, permissions: &Permissions) -> io::Result { - let file = std::fs::OpenOptions::new() - .write(true) - .create_new(true) - .open(path)?; - file.set_permissions(permissions.clone())?; - Ok(file) -} - -fn temp_path_for(path: &Path, operation: &str) -> PathBuf { - let mut file_name = path - .file_name() - .map(OsStr::to_os_string) - .unwrap_or_else(|| OsStr::new("rollout").to_os_string()); - let counter = TEMP_COUNTER.fetch_add(1, Ordering::Relaxed); - file_name.push(format!( - ".{operation}.{}.{counter}{TEMP_SUFFIX}", - std::process::id() - )); - path.with_file_name(file_name) -} - #[cfg(test)] #[path = "compression_tests.rs"] mod tests; diff --git a/codex-rs/rollout/src/compression/access_metadata.rs b/codex-rs/rollout/src/compression/access_metadata.rs new file mode 100644 index 0000000000..a34e74160d --- /dev/null +++ b/codex-rs/rollout/src/compression/access_metadata.rs @@ -0,0 +1,119 @@ +//! Preserves access to replacement rollout files before they are published. +//! Metadata failures leave the source intact; never silently replace ownership +//! or discard a macOS ACL when switching between plain and compressed files. +//! Apply ACLs last because they can prohibit subsequent metadata updates. +//! Reject delete-deny ACLs before installation so failed replacements remain removable. + +use std::fs::File; +use std::fs::FileTimes; +use std::io; +#[cfg(target_os = "macos")] +use std::os::fd::AsRawFd; +#[cfg(unix)] +use std::os::unix::fs::MetadataExt; + +// libc does not expose the macOS ACL APIs. +#[cfg(target_os = "macos")] +unsafe extern "C" { + fn acl_get_fd(fd: libc::c_int) -> *mut libc::c_void; + fn acl_get_entry( + acl: *mut libc::c_void, + entry_id: libc::c_int, + entry: *mut *mut libc::c_void, + ) -> libc::c_int; + fn acl_get_tag_type(entry: *mut libc::c_void, tag: *mut libc::c_uint) -> libc::c_int; + fn acl_get_permset_mask_np(entry: *mut libc::c_void, permissions: *mut u64) -> libc::c_int; + fn acl_init(count: libc::c_int) -> *mut libc::c_void; + fn acl_set_fd(fd: libc::c_int, acl: *mut libc::c_void) -> libc::c_int; + fn acl_free(acl: *mut libc::c_void) -> libc::c_int; +} + +pub(super) fn copy(source: &File, destination: &File) -> io::Result<()> { + let metadata = source.metadata()?; + #[cfg(unix)] + { + let destination_metadata = destination.metadata()?; + if (destination_metadata.uid(), destination_metadata.gid()) + != (metadata.uid(), metadata.gid()) + { + // Chown can clear mode bits, so restore permissions afterwards. + // Failure must abort publication rather than change who owns the file. + std::os::unix::fs::fchown(destination, Some(metadata.uid()), Some(metadata.gid()))?; + } + } + destination.set_permissions(metadata.permissions())?; + destination.set_times(FileTimes::new().set_modified(metadata.modified()?))?; + #[cfg(target_os = "macos")] + { + // COPYFILE_ACL merges inherited destination entries and drops inherited + // source entries. Replace the entire ACL to preserve its order and denies. + // SAFETY: source stays open and the returned ACL is freed below. + let mut acl = unsafe { acl_get_fd(source.as_raw_fd()) }; + let source_has_no_acl = acl.is_null(); + if source_has_no_acl { + let error = io::Error::last_os_error(); + if error.raw_os_error() != Some(libc::ENOENT) { + return Err(error); + } + // No source ACL: also clear any ACL inherited by the new file. + // SAFETY: zero is a valid entry count; allocation failure returns null. + acl = unsafe { + acl_init(/*count*/ 0) + }; + if acl.is_null() { + return Err(io::Error::last_os_error()); + } + } + let result = (|| { + // These values come from macOS . + const ACL_EXTENDED_DENY: libc::c_uint = 2; + const ACL_DELETE: u64 = 1 << 4; + let mut entry_id = 0; + loop { + let mut entry = std::ptr::null_mut(); + // SAFETY: acl is valid and entry points to writable pointer storage. + if unsafe { acl_get_entry(acl, entry_id, &mut entry) } != 0 { + let error = io::Error::last_os_error(); + // macOS returns EINVAL when the entry index reaches the end. + if error.raw_os_error() == Some(libc::EINVAL) { + break; + } + return Err(error); + } + let mut tag = 0; + let mut permissions = 0; + // SAFETY: entry belongs to the live acl; both outputs are writable. + if unsafe { acl_get_tag_type(entry, &mut tag) } != 0 + || unsafe { acl_get_permset_mask_np(entry, &mut permissions) } != 0 + { + return Err(io::Error::last_os_error()); + } + // Conservatively reject any delete deny, regardless of principal or + // ordering. Installing it can prevent both publication and cleanup, + // and the ACL may also deny restoring permissions after a failure. + if tag == ACL_EXTENDED_DENY && permissions & ACL_DELETE != 0 { + return Err(io::Error::new( + io::ErrorKind::PermissionDenied, + "cannot replace a rollout with a delete-deny ACL", + )); + } + entry_id += 1; + } + // SAFETY: destination stays open and acl is a valid owned allocation. + if unsafe { acl_set_fd(destination.as_raw_fd(), acl) } == 0 { + return Ok(()); + } + let error = io::Error::last_os_error(); + // Filesystems without ACL support reject even an empty ACL. There + // is nothing to preserve only when the source had no ACL. + if source_has_no_acl && error.raw_os_error() == Some(libc::ENOTSUP) { + return Ok(()); + } + Err(error) + })(); + // SAFETY: acl came from acl_get_fd or acl_init and is no longer used. + unsafe { acl_free(acl) }; + result?; + } + Ok(()) +} diff --git a/codex-rs/rollout/src/compression_tests.rs b/codex-rs/rollout/src/compression_tests.rs index 22e1d66744..93b9e48c6a 100644 --- a/codex-rs/rollout/src/compression_tests.rs +++ b/codex-rs/rollout/src/compression_tests.rs @@ -3,6 +3,7 @@ use std::fs; use std::fs::FileTimes; #[cfg(unix)] use std::os::unix::fs::PermissionsExt; +use std::sync::atomic::Ordering; use std::time::Duration; use std::time::SystemTime; @@ -441,10 +442,21 @@ async fn worker_compresses_old_active_and_archived_rollouts() -> anyhow::Result< let stale_temp = active_path.with_file_name("rollout-stale.jsonl.zst.tmp"); fs::write(&stale_temp, "stale temp")?; set_old_mtime(&stale_temp)?; + // A crash during metadata restoration can leave the data file unreadable. + #[cfg(unix)] + fs::set_permissions(&stale_temp, fs::Permissions::from_mode(0o000))?; let fresh_temp = active_path.with_file_name("rollout-fresh.jsonl.zst.tmp"); fs::write(&fresh_temp, "fresh temp")?; + // Pause materialization after restoring the old metadata, before publication. + let mut live_temp = tempfile::Builder::new() + .prefix(".rollout-materialize-") + .tempfile_in(active_path.parent().unwrap())?; + live_temp.write_all(b"live materialization")?; + access_metadata::copy(&fs::File::open(&active_path)?, live_temp.as_file())?; + let source_modified = fs::metadata(&active_path)?.modified()?; + worker::run(home.path().to_path_buf(), Startup).await?; assert!(!active_path.exists()); @@ -455,6 +467,13 @@ async fn worker_compresses_old_active_and_archived_rollouts() -> anyhow::Result< assert!(!compressed_rollout_path(&fresh_path).exists()); assert!(!stale_temp.exists()); assert!(fresh_temp.exists()); + let materialized_path = home.path().join("materialized.jsonl"); + persist_temp_file_noclobber(live_temp, &materialized_path)?; + assert_eq!( + fs::metadata(&materialized_path)?.modified()?, + source_modified + ); + assert_eq!(fs::read(&materialized_path)?, b"live materialization"); assert!( home.path() .join(".tmp") @@ -623,6 +642,211 @@ async fn compression_preserves_rollout_permissions() -> anyhow::Result<()> { Ok(()) } +#[cfg(target_os = "macos")] +#[tokio::test] +async fn worker_preserves_inherited_acl_through_append() -> anyhow::Result<()> { + use anyhow::Context; + use std::os::unix::fs::MetadataExt; + use std::process::Command; + + let home = TempDir::new()?; + let uuid = Uuid::from_u128(61); + let thread_id = ThreadId::from_string(&uuid.to_string())?; + let path = archived_rollout_path(home.path(), "2025-01-03T12-00-00", uuid); + write_rollout(&path, thread_id, "ACL-protected transcript")?; + set_old_mtime(&path)?; + // An archived rollout can retain inherited denies from its former directory. + let acl = Command::new("/bin/chmod") + .args(["+ai#", "0", "everyone deny writeattr"]) + .arg(&path) + .output()?; + assert!(acl.status.success(), "{acl:?}"); + fs::set_permissions(&path, fs::Permissions::from_mode(0o000))?; + let acl = Command::new("/bin/chmod") + .args(["+a#", "1", "everyone allow read,write,append,writeattr"]) + .arg(&path) + .output()?; + assert!(acl.status.success(), "{acl:?}"); + let before = fs::metadata(&path)?; + let expected = RolloutRecorder::load_rollout_items(&path) + .await + .context("read the ACL-protected source")?; + fs::OpenOptions::new() + .append(true) + .open(&path) + .context("the source ACL must permit appending before compression")?; + assert_eq!( + fs::File::open(&path)? + .set_times(FileTimes::new().set_modified(SystemTime::now())) + .unwrap_err() + .kind(), + io::ErrorKind::PermissionDenied + ); + + // Exercise discovery, compression, final metadata and publication through + // the real worker, not the test-only compress_now helper. + worker::run(home.path().to_path_buf(), Startup).await?; + let compressed = compressed_rollout_path(&path); + assert!(!path.exists()); + assert_eq!( + fs::File::open(&compressed)? + .set_times(FileTimes::new().set_modified(SystemTime::now())) + .unwrap_err() + .kind(), + io::ErrorKind::PermissionDenied + ); + let after = fs::metadata(&compressed)?; + assert_eq!(after.modified()?, before.modified()?); + assert_eq!( + (after.uid(), after.gid(), after.permissions().mode()), + (before.uid(), before.gid(), before.permissions().mode()) + ); + assert_eq!( + serde_json::to_value( + RolloutRecorder::load_rollout_items(&path) + .await + .context("read the worker's compressed replacement")? + )?, + serde_json::to_value(&expected)? + ); + + let appended = RolloutItem::EventMsg(EventMsg::UserMessage(UserMessageEvent { + message: "append through the preserved ACL".to_string(), + ..Default::default() + })); + append_rollout_item_to_path(&path, &appended) + .await + .context("materialize and append through the preserved ACL")?; + assert!(!compressed.exists()); + assert_eq!( + fs::File::open(&path)? + .set_times(FileTimes::new().set_modified(SystemTime::now())) + .unwrap_err() + .kind(), + io::ErrorKind::PermissionDenied + ); + let after = fs::metadata(&path)?; + assert_eq!( + (after.uid(), after.gid(), after.permissions().mode()), + (before.uid(), before.gid(), before.permissions().mode()) + ); + let mut expected_items = expected.0; + expected_items.push(appended); + assert_eq!( + serde_json::to_value(RolloutRecorder::load_rollout_items(&path).await?)?, + serde_json::to_value((expected_items, expected.1, expected.2))? + ); + Ok(()) +} + +#[cfg(unix)] +#[tokio::test] +async fn worker_preserves_alternate_group_through_materialization() -> anyhow::Result<()> { + use std::os::unix::fs::MetadataExt; + use std::process::Command; + + let home = TempDir::new()?; + let uuid = Uuid::from_u128(62); + let thread_id = ThreadId::from_string(&uuid.to_string())?; + let path = archived_rollout_path(home.path(), "2025-01-03T12-00-00", uuid); + write_rollout(&path, thread_id, "different group")?; + set_old_mtime(&path)?; + let original_group = fs::metadata(&path)?.gid(); + let groups = Command::new("id").arg("-G").output()?; + assert!(groups.status.success()); + let Some(group) = std::str::from_utf8(&groups.stdout)? + .split_whitespace() + .map(str::parse::) + .collect::, _>>()? + .into_iter() + .find(|group| *group != original_group) + else { + eprintln!("skipping ownership regression: no alternate supplementary group"); + return Ok(()); + }; + std::os::unix::fs::chown(&path, /*uid*/ None, Some(group))?; + assert_eq!(fs::metadata(&path)?.gid(), group); + + worker::run(home.path().to_path_buf(), Startup).await?; + assert!(!path.exists()); + assert_eq!(fs::metadata(compressed_rollout_path(&path))?.gid(), group); + materialize_rollout_for_append(&path, /*writer_lock*/ None).await?; + assert_eq!(fs::metadata(&path)?.gid(), group); + Ok(()) +} + +#[cfg(target_os = "macos")] +#[tokio::test] +async fn worker_does_not_leak_temps_for_deletion_protected_rollouts() -> anyhow::Result<()> { + use std::process::Command; + + let home = TempDir::new()?; + let uuid = Uuid::from_u128(63); + let thread_id = ThreadId::from_string(&uuid.to_string())?; + let path = archived_rollout_path(home.path(), "2025-01-03T12-00-00", uuid); + write_rollout(&path, thread_id, "deletion-protected transcript")?; + set_old_mtime(&path)?; + let acl = Command::new("/bin/chmod") + .args(["+a", "everyone deny delete,writesecurity"]) + .arg(&path) + .output()?; + assert!(acl.status.success(), "{acl:?}"); + let contents = fs::read(&path)?; + let modified = fs::metadata(&path)?.modified()?; + let access = Command::new("/bin/ls").arg("-le").arg(&path).output()?; + assert!(access.status.success(), "{access:?}"); + + // Collect results before asserting so even a regression's protected temps + // have their ACLs cleared before TempDir tries to remove them. + let result: anyhow::Result<_> = async { + let mut files_after_passes = Vec::new(); + for _ in 0..2 { + worker::run(home.path().to_path_buf(), Startup).await?; + files_after_passes.push( + fs::read_dir(path.parent().unwrap())? + .map(|entry| entry.map(|entry| entry.path())) + .collect::>>()?, + ); + // Allow the next pass without waiting for the run marker to expire. + fs::remove_file(home.path().join(".tmp/rollout-compression.lock"))?; + } + let access_after = Command::new("/bin/ls").arg("-le").arg(&path).output()?; + Ok(( + files_after_passes, + fs::read(&path)?, + fs::metadata(&path)?.modified()?, + access_after, + fs::remove_file(&path).map_err(|err| err.kind()), + )) + } + .await; + for entry in fs::read_dir(path.parent().unwrap())? { + let cleanup = Command::new("/bin/chmod") + .arg("-N") + .arg(entry?.path()) + .output()?; + assert!(cleanup.status.success(), "{cleanup:?}"); + } + let (files_after_passes, contents_after, modified_after, access_after, deletion) = result?; + assert!(access_after.status.success(), "{access_after:?}"); + assert_eq!(files_after_passes, vec![vec![path.clone()], vec![path]]); + assert_eq!( + ( + contents_after, + modified_after, + access_after.stdout, + deletion + ), + ( + contents, + modified, + access.stdout, + Err(io::ErrorKind::PermissionDenied) + ) + ); + Ok(()) +} + #[cfg(unix)] #[tokio::test] async fn append_materialization_preserves_compressed_rollout_permissions() -> anyhow::Result<()> { @@ -656,11 +880,12 @@ async fn append_materialization_preserves_compressed_rollout_permissions() -> an #[test] fn persist_temp_file_noclobber_installs_completed_temp() -> anyhow::Result<()> { let home = TempDir::new()?; - let temp_path = home.path().join("rollout.jsonl.tmp"); + let mut temp_file = tempfile::NamedTempFile::new_in(home.path())?; + let temp_path = temp_file.path().to_path_buf(); let destination = home.path().join("rollout.jsonl"); - fs::write(&temp_path, "completed rollout")?; + temp_file.write_all(b"completed rollout")?; - persist_temp_file_noclobber(&temp_path, &destination)?; + persist_temp_file_noclobber(temp_file, &destination)?; assert!(!temp_path.exists()); assert_eq!(fs::read_to_string(destination)?, "completed rollout"); @@ -670,12 +895,13 @@ fn persist_temp_file_noclobber_installs_completed_temp() -> anyhow::Result<()> { #[test] fn persist_temp_file_noclobber_does_not_replace_existing_destination() -> anyhow::Result<()> { let home = TempDir::new()?; - let temp_path = home.path().join("rollout.jsonl.tmp"); + let mut temp_file = tempfile::NamedTempFile::new_in(home.path())?; + let temp_path = temp_file.path().to_path_buf(); let destination = home.path().join("rollout.jsonl"); - fs::write(&temp_path, "candidate rollout")?; + temp_file.write_all(b"candidate rollout")?; fs::write(&destination, "existing rollout")?; - persist_temp_file_noclobber(&temp_path, &destination)?; + persist_temp_file_noclobber(temp_file, &destination)?; assert!(!temp_path.exists()); assert_eq!(fs::read_to_string(destination)?, "existing rollout"); @@ -693,6 +919,18 @@ async fn compression_preserves_read_only_rollout_permissions() -> anyhow::Result set_old_mtime(&rollout_path)?; fs::set_permissions(&rollout_path, fs::Permissions::from_mode(0o400))?; let source_modified = fs::metadata(&rollout_path)?.modified()?; + #[cfg(target_os = "macos")] + { + // New files inherit an allow that is absent from the existing source. + let acl = std::process::Command::new("/bin/chmod") + .args([ + "+a", + "everyone allow write,append,file_inherit,only_inherit", + ]) + .arg(rollout_path.parent().unwrap()) + .output()?; + assert!(acl.status.success(), "{acl:?}"); + } worker::run(home.path().to_path_buf(), Startup).await?; @@ -701,6 +939,27 @@ async fn compression_preserves_read_only_rollout_permissions() -> anyhow::Result assert!(!rollout_path.exists()); assert_eq!(compressed_metadata.permissions().mode() & 0o777, 0o400); assert_eq!(compressed_metadata.modified()?, source_modified); + #[cfg(target_os = "macos")] + { + assert_eq!( + fs::OpenOptions::new() + .append(true) + .open(&compressed_path) + .unwrap_err() + .kind(), + io::ErrorKind::PermissionDenied + ); + crate::materialize_rollout_for_reference(&rollout_path).await?; + assert_eq!(fs::metadata(&rollout_path)?.modified()?, source_modified); + assert_eq!( + fs::OpenOptions::new() + .append(true) + .open(&rollout_path) + .unwrap_err() + .kind(), + io::ErrorKind::PermissionDenied + ); + } Ok(()) } diff --git a/codex-rs/rollout/src/recorder.rs b/codex-rs/rollout/src/recorder.rs index 765d83ccc7..4dbd9e5fe1 100644 --- a/codex-rs/rollout/src/recorder.rs +++ b/codex-rs/rollout/src/recorder.rs @@ -1776,12 +1776,22 @@ fn open_log_file(path: &Path) -> std::io::Result { .create(true) .open(path)?; if refresh_modified_time { - file.set_times(std::fs::FileTimes::new().set_modified(std::time::SystemTime::now()))?; + refresh_rollout_modified_time(&file)?; } ensure_rollout_is_newline_terminated(&mut file)?; Ok(file) } +fn refresh_rollout_modified_time(file: &File) -> std::io::Result<()> { + match file.set_times(std::fs::FileTimes::new().set_modified(std::time::SystemTime::now())) { + Ok(()) => Ok(()), + // A preserved ACL can permit appending while denying explicit timestamp changes. + // The append itself will update mtime, so this refresh must not prevent it. + Err(err) if err.kind() == std::io::ErrorKind::PermissionDenied => Ok(()), + Err(err) => Err(err), + } +} + /// Mutable state owned by the background rollout writer. /// /// Items are first appended to `pending_items`; persist/flush/shutdown remove each item from that @@ -2045,7 +2055,7 @@ async fn open_rollout_for_append( .append(true) .open(path_for_open.as_path())?; if refresh_modified_time { - file.set_times(std::fs::FileTimes::new().set_modified(std::time::SystemTime::now()))?; + refresh_rollout_modified_time(&file)?; } ensure_rollout_is_newline_terminated(&mut file)?; let ordinal_state = ordinal_state_for_rollout(&mut file, path_for_open.as_path())?; diff --git a/codex-rs/rollout/src/recorder_tests.rs b/codex-rs/rollout/src/recorder_tests.rs index d8916fea98..e440aa1617 100644 --- a/codex-rs/rollout/src/recorder_tests.rs +++ b/codex-rs/rollout/src/recorder_tests.rs @@ -174,6 +174,31 @@ async fn opening_existing_rollout_preserves_modified_time() -> std::io::Result<( Ok(()) } +#[cfg(target_os = "macos")] +#[test] +fn open_log_file_materializes_rollout_when_acl_denies_timestamp_changes() -> anyhow::Result<()> { + let home = TempDir::new()?; + let rollout_path = home.path().join("rollout.jsonl"); + let compressed_path = rollout_path.with_extension("jsonl.zst"); + fs::write( + &compressed_path, + zstd::stream::encode_all(&b"original\n"[..], 0)?, + )?; + let acl = std::process::Command::new("/bin/chmod") + .args(["+a", "everyone deny writeattr"]) + .arg(&compressed_path) + .output()?; + assert!(acl.status.success(), "{acl:?}"); + + let mut file = open_log_file(&rollout_path)?; + file.write_all(b"appended\n")?; + drop(file); + + assert!(!compressed_path.exists()); + assert_eq!(fs::read(&rollout_path)?, b"original\nappended\n"); + Ok(()) +} + #[tokio::test] async fn state_db_init_backfills_before_returning() -> anyhow::Result<()> { let home = TempDir::new().expect("temp dir");