pharos_runtime_core/
fs.rs

1use std::fs::{self, OpenOptions};
2use std::io::{self, Read, Write};
3use std::path::{Path, PathBuf};
4use std::sync::atomic::{AtomicU64, Ordering};
5use std::time::{SystemTime, UNIX_EPOCH};
6
7/// Default private runtime-state mode on Unix hosts.
8///
9/// # Examples
10///
11/// ```rust,ignore
12/// let _ = pharos_runtime_core::fs::PRIVATE_FILE_MODE;
13/// ```
14///
15/// # Related
16///
17/// - [`crate::fs`] for the surrounding module surface.
18pub const PRIVATE_FILE_MODE: u32 = 0o640;
19
20static TEMP_FILE_COUNTER: AtomicU64 = AtomicU64::new(0);
21
22/// Ensures directory.
23///
24/// # Parameters
25///
26/// - `path`: Input consumed by [`crate::fs::ensure_directory`].
27///
28/// # Returns
29///
30/// Returns the computed value produced by [`crate::fs::ensure_directory`].
31///
32/// # Examples
33///
34/// ```rust,ignore
35/// let _result = pharos_runtime_core::fs::ensure_directory(std::path::Path::new("fixture.json"));
36/// ```
37///
38/// # Related
39///
40/// - [`crate::fs`] for adjacent helpers that share the same runtime surface.
41pub fn ensure_directory(path: &Path) -> i32 {
42    if path.as_os_str().is_empty() || path.is_dir() {
43        return 0;
44    }
45    fs::create_dir_all(path).map(|_| 0).unwrap_or(69)
46}
47
48/// Builds the temporary path.
49fn temporary_path(path: &Path, attempt: u64) -> Option<PathBuf> {
50    let parent = path.parent()?;
51    let name = path.file_name()?.to_string_lossy();
52    let timestamp = SystemTime::now()
53        .duration_since(UNIX_EPOCH)
54        .ok()?
55        .as_nanos();
56    let counter = TEMP_FILE_COUNTER.fetch_add(1, Ordering::Relaxed);
57    Some(parent.join(format!(
58        ".{name}.pharos-tmp-{}-{timestamp}-{counter}-{attempt}",
59        std::process::id()
60    )))
61}
62
63/// Performs sync directory.
64fn sync_directory(path: &Path) -> io::Result<()> {
65    #[cfg(unix)]
66    {
67        OpenOptions::new().read(true).open(path)?.sync_all()
68    }
69    #[cfg(not(unix))]
70    {
71        let _ = path;
72        Ok(())
73    }
74}
75
76/// Replaces a file through a same-directory durable temporary file.
77fn atomic_write(path: &Path, content: &[u8], mode: u32) -> io::Result<()> {
78    #[cfg(not(unix))]
79    let _ = mode;
80    let Some(parent) = path.parent() else {
81        return Err(io::Error::new(
82            io::ErrorKind::InvalidInput,
83            "file path has no parent",
84        ));
85    };
86    for attempt in 0..8 {
87        let Some(temporary) = temporary_path(path, attempt) else {
88            return Err(io::Error::new(
89                io::ErrorKind::InvalidInput,
90                "file path has no filename",
91            ));
92        };
93        let mut options = OpenOptions::new();
94        options.create_new(true).write(true);
95        #[cfg(unix)]
96        {
97            use std::os::unix::fs::OpenOptionsExt;
98            options.mode(mode);
99        }
100        let mut file = match options.open(&temporary) {
101            Ok(file) => file,
102            Err(error) if error.kind() == io::ErrorKind::AlreadyExists => continue,
103            Err(error) => return Err(error),
104        };
105        let result = (|| {
106            file.write_all(content)?;
107            file.sync_all()?;
108            drop(file);
109            fs::rename(&temporary, path)?;
110            sync_directory(parent)
111        })();
112        if result.is_err() {
113            let _ = fs::remove_file(&temporary);
114        }
115        return result;
116    }
117    Err(io::Error::new(
118        io::ErrorKind::AlreadyExists,
119        "could not allocate a unique temporary file",
120    ))
121}
122
123/// Resolves a write destination below a trusted root without following a final symlink.
124fn validate_write_destination(root: &Path, path: &Path) -> io::Result<()> {
125    let canonical_root = fs::canonicalize(root)?;
126    let parent = path
127        .parent()
128        .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidInput, "file path has no parent"))?;
129    let canonical_parent = fs::canonicalize(parent)?;
130    if !canonical_parent.starts_with(&canonical_root) {
131        return Err(io::Error::new(
132            io::ErrorKind::PermissionDenied,
133            "write destination escapes its capability root",
134        ));
135    }
136    match fs::symlink_metadata(path) {
137        Ok(metadata) if metadata.file_type().is_symlink() || !metadata.is_file() => Err(
138            io::Error::new(io::ErrorKind::PermissionDenied, "unsafe write destination"),
139        ),
140        Ok(_) => Ok(()),
141        Err(error) if error.kind() == io::ErrorKind::NotFound => Ok(()),
142        Err(error) => Err(error),
143    }
144}
145
146/// Writes bytes below an explicit capability root through durable atomic replacement.
147///
148/// # Parameters
149///
150/// - `root`: Input consumed by [`crate::fs::write_file_within`].
151/// - `path`: Input consumed by [`crate::fs::write_file_within`].
152/// - `content`: Input consumed by [`crate::fs::write_file_within`].
153/// - `mode`: Input consumed by [`crate::fs::write_file_within`].
154///
155/// # Returns
156///
157/// Returns the computed value produced by [`crate::fs::write_file_within`].
158///
159/// # Examples
160///
161/// ```rust,ignore
162/// let _result = pharos_runtime_core::fs::write_file_within(std::path::Path::new("."), std::path::Path::new("fixture.json"), todo!(), 0o640);
163/// ```
164///
165/// # Related
166///
167/// - [`crate::fs`] for adjacent helpers that share the same runtime surface.
168pub fn write_file_within(root: &Path, path: &Path, content: &[u8], mode: u32) -> io::Result<()> {
169    if !(mode == 0o600 || mode == 0o640 || mode == 0o644) {
170        return Err(io::Error::new(
171            io::ErrorKind::InvalidInput,
172            "unsupported file mode",
173        ));
174    }
175    let parent = path
176        .parent()
177        .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidInput, "file path has no parent"))?;
178    fs::create_dir_all(parent)?;
179    validate_write_destination(root, path)?;
180    atomic_write(path, content, mode)?;
181    #[cfg(unix)]
182    {
183        use std::os::unix::fs::PermissionsExt;
184        fs::set_permissions(path, fs::Permissions::from_mode(mode))?;
185    }
186    Ok(())
187}
188
189/// Appends bytes to a regular file below an explicit capability root.
190///
191/// # Parameters
192///
193/// - `root`: Input consumed by [`crate::fs::append_file_within`].
194/// - `path`: Input consumed by [`crate::fs::append_file_within`].
195/// - `content`: Input consumed by [`crate::fs::append_file_within`].
196/// - `mode`: Input consumed by [`crate::fs::append_file_within`].
197///
198/// # Returns
199///
200/// Returns the computed value produced by [`crate::fs::append_file_within`].
201///
202/// # Examples
203///
204/// ```rust,ignore
205/// let _result = pharos_runtime_core::fs::append_file_within(std::path::Path::new("."), std::path::Path::new("fixture.json"), todo!(), 0o640);
206/// ```
207///
208/// # Related
209///
210/// - [`crate::fs`] for adjacent helpers that share the same runtime surface.
211pub fn append_file_within(root: &Path, path: &Path, content: &[u8], mode: u32) -> io::Result<()> {
212    if !(mode == 0o600 || mode == 0o640 || mode == 0o644) {
213        return Err(io::Error::new(
214            io::ErrorKind::InvalidInput,
215            "unsupported file mode",
216        ));
217    }
218    let parent = path
219        .parent()
220        .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidInput, "file path has no parent"))?;
221    fs::create_dir_all(parent)?;
222    validate_write_destination(root, path)?;
223    let mut options = OpenOptions::new();
224    options.create(true).append(true);
225    #[cfg(unix)]
226    {
227        use std::os::unix::fs::OpenOptionsExt;
228        options.mode(mode);
229    }
230    let mut file = options.open(path)?;
231    file.write_all(content)?;
232    file.sync_data()?;
233    #[cfg(unix)]
234    {
235        use std::os::unix::fs::PermissionsExt;
236        fs::set_permissions(path, fs::Permissions::from_mode(mode))?;
237    }
238    Ok(())
239}
240
241/// Writes a local JSONL outbox delivery once across retries and processes.
242/// A separate OS-locked companion survives atomic destination replacement.
243/// Both reading and replacement are bounded to 64 MiB; incomplete records fail
244/// closed instead of losing existing output or duplicating a delivery.
245pub fn append_json_record_once(
246    root: &Path,
247    path: &Path,
248    key: &str,
249    record: &serde_json::Value,
250    mode: u32,
251) -> io::Result<()> {
252    const LIMIT: usize = 64 * 1024 * 1024;
253    if key.is_empty()
254        || key.len() > 256
255        || !record.is_object()
256        || !matches!(mode, 0o600 | 0o640 | 0o644)
257    {
258        return Err(io::Error::new(
259            io::ErrorKind::InvalidInput,
260            "invalid local delivery",
261        ));
262    }
263    let parent = path
264        .parent()
265        .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidInput, "missing delivery parent"))?;
266    fs::create_dir_all(parent)?;
267    validate_write_destination(root, path)?;
268    let name = path
269        .file_name()
270        .and_then(|name| name.to_str())
271        .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidInput, "invalid delivery filename"))?;
272    let lock_path = parent.join(format!(".{name}.delivery-lock"));
273    validate_write_destination(root, &lock_path)?;
274    let mut options = OpenOptions::new();
275    options.read(true).write(true).create(true).truncate(false);
276    #[cfg(unix)]
277    {
278        use std::os::unix::fs::OpenOptionsExt;
279        options.mode(0o600);
280    }
281    let lock = options.open(lock_path)?;
282    let started = std::time::Instant::now();
283    while lock.try_lock().is_err() {
284        if started.elapsed() >= std::time::Duration::from_millis(250) {
285            return Err(io::Error::new(
286                io::ErrorKind::WouldBlock,
287                "local delivery lock unavailable",
288            ));
289        }
290        std::thread::sleep(std::time::Duration::from_millis(5));
291    }
292    validate_write_destination(root, path)?;
293    let mut bytes = Vec::new();
294    match fs::File::open(path) {
295        Ok(file) => {
296            file.take((LIMIT + 1) as u64).read_to_end(&mut bytes)?;
297        }
298        Err(error) if error.kind() == io::ErrorKind::NotFound => {}
299        Err(error) => return Err(error),
300    }
301    if bytes.len() > LIMIT || (!bytes.is_empty() && bytes.last() != Some(&b'\n')) {
302        return Err(io::Error::new(
303            io::ErrorKind::InvalidData,
304            "invalid or oversized delivery output",
305        ));
306    }
307    for line in bytes
308        .split(|byte| *byte == b'\n')
309        .filter(|line| !line.is_empty())
310    {
311        let previous: serde_json::Value = serde_json::from_slice(line)
312            .map_err(|_| io::Error::new(io::ErrorKind::InvalidData, "invalid delivery record"))?;
313        if previous
314            .get("_pharos_delivery_key")
315            .and_then(serde_json::Value::as_str)
316            == Some(key)
317        {
318            return Ok(());
319        }
320    }
321    let mut value = record.clone();
322    value
323        .as_object_mut()
324        .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidInput, "invalid delivery record"))?
325        .insert(
326            "_pharos_delivery_key".to_owned(),
327            serde_json::Value::String(key.to_owned()),
328        );
329    let encoded = serde_json::to_vec(&value)
330        .map_err(|_| io::Error::new(io::ErrorKind::InvalidInput, "invalid delivery record"))?;
331    if bytes.len().saturating_add(encoded.len()).saturating_add(1) > LIMIT {
332        return Err(io::Error::new(
333            io::ErrorKind::InvalidData,
334            "delivery output exceeds limit",
335        ));
336    }
337    bytes.extend(encoded);
338    bytes.push(b'\n');
339    atomic_write(path, &bytes, mode)
340}
341
342/// Writes bytes using the destination directory as the least-authority capability root.
343///
344/// # Parameters
345///
346/// - `path`: Input consumed by [`crate::fs::write_binary_file`].
347/// - `content`: Input consumed by [`crate::fs::write_binary_file`].
348/// - `mode`: Input consumed by [`crate::fs::write_binary_file`].
349///
350/// # Returns
351///
352/// Returns the computed value produced by [`crate::fs::write_binary_file`].
353///
354/// # Examples
355///
356/// ```rust,ignore
357/// let _result = pharos_runtime_core::fs::write_binary_file(std::path::Path::new("fixture.json"), todo!(), 0o640);
358/// ```
359///
360/// # Related
361///
362/// - [`crate::fs`] for adjacent helpers that share the same runtime surface.
363pub fn write_binary_file(path: &Path, content: &[u8], mode: u32) -> io::Result<()> {
364    let parent = path
365        .parent()
366        .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidInput, "file path has no parent"))?;
367    fs::create_dir_all(parent)?;
368    write_file_within(parent, path, content, mode)
369}
370
371/// Writes a text document through durable atomic replacement.
372///
373/// # Parameters
374///
375/// - `path`: Input consumed by [`crate::fs::write_text_file`].
376/// - `content`: Input consumed by [`crate::fs::write_text_file`].
377///
378/// # Returns
379///
380/// Returns the computed value produced by [`crate::fs::write_text_file`].
381///
382/// # Examples
383///
384/// ```rust,ignore
385/// let _result = pharos_runtime_core::fs::write_text_file(std::path::Path::new("fixture.json"), todo!());
386/// ```
387///
388/// # Related
389///
390/// - [`crate::fs`] for adjacent helpers that share the same runtime surface.
391pub fn write_text_file(path: &Path, content: &str) -> i32 {
392    let Some(parent) = path.parent() else {
393        return 69;
394    };
395    if ensure_directory(parent) != 0 {
396        return 69;
397    }
398    write_file_within(parent, path, content.as_bytes(), PRIVATE_FILE_MODE)
399        .map(|_| 0)
400        .unwrap_or(69)
401}
402
403#[cfg(test)]
404mod tests {
405    use super::{write_file_within, write_text_file};
406    use std::fs;
407    use std::time::{SystemTime, UNIX_EPOCH};
408
409    /// Concurrent replays append exactly one durable record per delivery key.
410    #[test]
411    fn local_outbox_replay_is_idempotent_across_writers() {
412        let directory = std::env::temp_dir().join(format!(
413            "pharos-outbox-{}-{}",
414            std::process::id(),
415            SystemTime::now()
416                .duration_since(UNIX_EPOCH)
417                .expect("clock")
418                .as_nanos()
419        ));
420        fs::create_dir_all(&directory).expect("outbox root");
421        let path = directory.join("events.jsonl");
422        std::thread::scope(|scope| {
423            for _ in 0..4 {
424                let directory = &directory;
425                let path = &path;
426                scope.spawn(move || {
427                    super::append_json_record_once(
428                        directory,
429                        path,
430                        "action-1",
431                        &serde_json::json!({"event":"committed"}),
432                        0o640,
433                    )
434                    .expect("replayed delivery");
435                });
436            }
437        });
438        let bytes = fs::read_to_string(&path).expect("outbox output");
439        assert_eq!(bytes.lines().count(), 1);
440        fs::write(&path, b"incomplete").expect("interrupted foreign write");
441        assert!(
442            super::append_json_record_once(
443                &directory,
444                &path,
445                "action-2",
446                &serde_json::json!({"event":"next"}),
447                0o640
448            )
449            .is_err()
450        );
451        assert_eq!(
452            fs::read(&path).expect("preserved invalid output"),
453            b"incomplete"
454        );
455        fs::remove_dir_all(directory).expect("cleanup outbox root");
456    }
457
458    #[test]
459    /// Writes text file replaces content without leaving temp files.
460    fn write_text_file_replaces_content_without_leaving_temp_files() {
461        let unique = SystemTime::now()
462            .duration_since(UNIX_EPOCH)
463            .expect("clock")
464            .as_nanos();
465        let directory = std::env::temp_dir().join(format!("pharos-fs-test-{unique}"));
466        fs::create_dir_all(&directory).expect("directory");
467        let path = directory.join("state.json");
468
469        assert_eq!(write_text_file(&path, "first"), 0);
470        assert_eq!(write_text_file(&path, "second"), 0);
471        assert_eq!(fs::read_to_string(&path).expect("content"), "second");
472        assert_eq!(fs::read_dir(&directory).expect("entries").count(), 1);
473
474        fs::remove_dir_all(directory).expect("cleanup");
475    }
476
477    #[cfg(unix)]
478    #[test]
479    /// Proves a configured symlink cannot redirect a capability-scoped write.
480    fn capability_write_rejects_symlink_destination() {
481        use std::os::unix::fs::symlink;
482
483        let unique = SystemTime::now()
484            .duration_since(UNIX_EPOCH)
485            .expect("clock")
486            .as_nanos();
487        let directory = std::env::temp_dir().join(format!("pharos-fs-symlink-test-{unique}"));
488        fs::create_dir_all(&directory).expect("directory");
489        let outside = std::env::temp_dir().join(format!("pharos-fs-outside-{unique}"));
490        fs::write(&outside, "preserved").expect("outside fixture");
491        let destination = directory.join("state.json");
492        symlink(&outside, &destination).expect("symlink fixture");
493
494        assert!(write_file_within(&directory, &destination, b"replaced", 0o640).is_err());
495        assert_eq!(
496            fs::read_to_string(&outside).expect("outside content"),
497            "preserved"
498        );
499
500        fs::remove_file(outside).expect("outside cleanup");
501        fs::remove_dir_all(directory).expect("cleanup");
502    }
503}
pharos_runtime_core/fs.rs

pharos_runtime_core/
fs.rs
