#![expect(
clippy::redundant_pub_crate,
reason = "explicit pub(crate) documents the crate-wide visibility intent at each item"
)]
use std::path::{Component, Path, PathBuf};
use std::sync::Arc;
use crate::error::{Error, Result};
use crate::sandbox::GrantSet;
#[derive(Debug, Clone)]
pub(crate) struct PathGuard {
grants: Arc<GrantSet>,
module: &'static str,
}
impl PathGuard {
pub(crate) const fn new(grants: Arc<GrantSet>, module: &'static str) -> Self {
Self { grants, module }
}
pub(crate) fn read(&self, operation: &'static str, raw: &str) -> Result<PathBuf> {
let resolved = Self::resolve(raw)?;
if self.grants.is_unrestricted() || self.grants.fs().allows_read(&resolved) {
return Ok(resolved);
}
Err(self.deny(operation, "read", &resolved))
}
pub(crate) fn write(&self, operation: &'static str, raw: &str) -> Result<PathBuf> {
let resolved = Self::resolve(raw)?;
if self.grants.is_unrestricted() || self.grants.fs().allows_write(&resolved) {
return Ok(resolved);
}
Err(self.deny(operation, "write", &resolved))
}
fn deny(&self, operation: &'static str, direction: &str, resolved: &Path) -> Error {
let roots = if direction == "read" {
self.grants.fs().read_roots()
} else {
self.grants.fs().write_roots()
};
let granted = if roots.is_empty() {
format!("no {direction} roots are granted")
} else {
let names: Vec<_> = roots.iter().map(|r| r.display().to_string()).collect();
format!("granted {direction} roots are {}", names.join(", "))
};
Error::Denied {
module: self.module,
operation,
detail: format!("`{}` is outside them — {granted}", resolved.display()),
}
}
fn resolve(raw: &str) -> Result<PathBuf> {
let absolute = std::path::absolute(raw).map_err(|_| Error::UncheckablePath {
path: raw.to_owned(),
reason: "the working directory could not be read",
})?;
let mut suffix: Vec<std::ffi::OsString> = Vec::new();
let mut probe = absolute;
loop {
if let Ok(canonical) = probe.canonicalize() {
let mut resolved = canonical;
for name in suffix.iter().rev() {
resolved.push(name);
}
return Ok(resolved);
}
match probe.components().next_back() {
Some(Component::Normal(name)) => suffix.push(name.to_owned()),
Some(Component::ParentDir) => {
return Err(Error::UncheckablePath {
path: raw.to_owned(),
reason: "it climbs through a directory that does not exist",
});
}
Some(Component::CurDir) => {}
_ => {
return Err(Error::UncheckablePath {
path: raw.to_owned(),
reason: "no part of it exists",
});
}
}
if !probe.pop() {
return Err(Error::UncheckablePath {
path: raw.to_owned(),
reason: "no part of it exists",
});
}
}
}
}
#[cfg(test)]
mod tests {
#![expect(
clippy::unwrap_used,
reason = "tests unwrap known-valid fixtures; a panic is the intended failure signal"
)]
use super::PathGuard;
use crate::sandbox::GrantSet;
use std::sync::Arc;
fn guard(build: impl FnOnce(GrantSet) -> GrantSet) -> PathGuard {
PathGuard::new(Arc::new(build(GrantSet::declared())), "fs")
}
#[test]
fn a_file_inside_a_read_root_is_permitted() {
let dir = tempfile::tempdir().unwrap();
let root = dir.path().canonicalize().unwrap();
std::fs::write(root.join("a.txt"), "x").unwrap();
let guard = guard(|g| g.with_fs(|fs| fs.read(&root)));
assert!(
guard
.read("read", root.join("a.txt").to_str().unwrap())
.is_ok()
);
}
#[test]
fn a_file_outside_every_read_root_is_refused() {
let dir = tempfile::tempdir().unwrap();
let outside = tempfile::tempdir().unwrap();
std::fs::write(outside.path().join("secret"), "x").unwrap();
let guard = guard(|g| g.with_fs(|fs| fs.read(dir.path())));
let err = guard
.read("read", outside.path().join("secret").to_str().unwrap())
.unwrap_err();
assert!(err.to_string().contains("fs.read denied"), "{err}");
}
#[test]
fn a_symlink_inside_the_root_pointing_out_of_it_is_refused() {
let outside = tempfile::tempdir().unwrap();
std::fs::write(outside.path().join("secret"), "leaked").unwrap();
let dir = tempfile::tempdir().unwrap();
let root = dir.path().canonicalize().unwrap();
std::os::unix::fs::symlink(outside.path().join("secret"), root.join("link")).unwrap();
let guard = guard(|g| g.with_fs(|fs| fs.read(&root)));
let err = guard
.read("read", root.join("link").to_str().unwrap())
.unwrap_err();
assert!(err.to_string().contains("denied"), "{err}");
}
#[test]
fn a_dotdot_through_a_symlink_does_not_escape() {
let outside = tempfile::tempdir().unwrap();
let outside_root = outside.path().canonicalize().unwrap();
std::fs::create_dir(outside_root.join("sub")).unwrap();
std::fs::write(outside_root.join("secret"), "leaked").unwrap();
let dir = tempfile::tempdir().unwrap();
let root = dir.path().canonicalize().unwrap();
std::os::unix::fs::symlink(outside_root.join("sub"), root.join("link")).unwrap();
let guard = guard(|g| g.with_fs(|fs| fs.read(&root)));
let attack = format!("{}/link/../secret", root.display());
assert!(
guard.read("read", &attack).is_err(),
"{attack} was permitted"
);
}
#[test]
fn a_path_that_does_not_exist_yet_is_checked_against_its_parent() {
let dir = tempfile::tempdir().unwrap();
let root = dir.path().canonicalize().unwrap();
let guard = guard(|g| g.with_fs(|fs| fs.write(&root)));
assert!(
guard
.write("write", root.join("new.txt").to_str().unwrap())
.is_ok()
);
assert!(
guard
.write("mkdir", root.join("a/b/c").to_str().unwrap())
.is_ok()
);
}
#[test]
fn a_dotdot_below_a_directory_that_does_not_exist_is_refused_rather_than_guessed() {
let dir = tempfile::tempdir().unwrap();
let root = dir.path().canonicalize().unwrap();
let guard = guard(|g| g.with_fs(|fs| fs.write(&root)));
let err = guard
.write("write", &format!("{}/absent/../ok.txt", root.display()))
.unwrap_err();
assert!(err.to_string().contains("cannot resolve"), "{err}");
}
#[test]
fn read_and_write_are_checked_against_their_own_roots() {
let dir = tempfile::tempdir().unwrap();
let root = dir.path().canonicalize().unwrap();
std::fs::write(root.join("a.txt"), "x").unwrap();
let guard = guard(|g| g.with_fs(|fs| fs.read(&root)));
let target = root.join("a.txt");
assert!(guard.read("read", target.to_str().unwrap()).is_ok());
let err = guard.write("write", target.to_str().unwrap()).unwrap_err();
assert!(err.to_string().contains("fs.write denied"), "{err}");
}
#[test]
fn an_unrestricted_policy_checks_nothing() {
let guard = PathGuard::new(Arc::new(GrantSet::unrestricted()), "fs");
assert!(guard.read("read", "/etc/hostname").is_ok());
}
#[test]
fn a_refusal_names_the_roots_that_were_granted() {
let dir = tempfile::tempdir().unwrap();
let root = dir.path().canonicalize().unwrap();
std::fs::write(root.join("a.txt"), "x").unwrap();
let guard = guard(|g| g.with_fs(|fs| fs.read(&root)));
let err = guard.read("read", "/etc/hostname").unwrap_err();
assert!(
err.to_string().contains(&root.display().to_string()),
"the refusal should say what was granted: {err}"
);
}
#[test]
fn a_refusal_with_no_roots_at_all_says_so() {
let guard = guard(|g| g);
let err = guard.read("read", "/etc/hostname").unwrap_err();
assert!(
err.to_string().contains("no read roots are granted"),
"{err}"
);
}
}