use std::path::{Component, Path, PathBuf};
pub(crate) fn is_blank_text(text: &str) -> bool {
text.chars().all(|c| {
c.is_whitespace() || matches!(c, '\u{200b}' | '\u{200c}' | '\u{200d}' | '\u{feff}')
})
}
pub fn is_blank_path(path: &str) -> bool {
is_blank_text(path)
}
pub fn safe_canonicalize(path: &Path) -> std::io::Result<PathBuf> {
dunce::canonicalize(path)
}
fn path_is_under(path: &Path, root: &Path) -> bool {
if path.starts_with(root) {
return true;
}
#[cfg(windows)]
{
if let Some(mapped) = windows_local_drive_share_path(path) {
if mapped.starts_with(root) {
return true;
}
if let Some(root_mapped) = windows_local_drive_share_path(root) {
return mapped.starts_with(root_mapped);
}
}
if let Some(drive) = windows_extended_prefix_to_drive(path) {
if drive.starts_with(root) {
return true;
}
let root_simple = dunce::simplified(root);
if drive.starts_with(root_simple) {
return true;
}
}
}
false
}
fn prefer_local_drive_path(path: PathBuf) -> PathBuf {
#[cfg(windows)]
{
if let Some(mapped) = windows_local_drive_share_path(&path) {
return safe_canonicalize(&mapped).unwrap_or(mapped);
}
}
path
}
pub(crate) fn prefer_openable_path(path: &Path) -> PathBuf {
#[cfg(windows)]
{
if let Some(drive) = windows_extended_prefix_to_drive(path) {
return drive;
}
if let Some(mapped) = windows_local_drive_share_path(path) {
return mapped;
}
}
path.to_path_buf()
}
#[cfg(windows)]
fn windows_extended_prefix_to_drive(path: &Path) -> Option<PathBuf> {
let raw = path.to_string_lossy();
let rest = raw
.strip_prefix("//?/")
.or_else(|| raw.strip_prefix(r"\\?\"))
.or_else(|| raw.strip_prefix("//./"))
.or_else(|| raw.strip_prefix(r"\\.\"))?;
if rest.len() >= 3
&& rest.as_bytes()[1] == b':'
&& (rest.as_bytes()[2] == b'/' || rest.as_bytes()[2] == b'\\')
{
return Some(PathBuf::from(rest.replace('/', "\\")));
}
None
}
#[cfg(windows)]
fn windows_local_drive_share_path(path: &Path) -> Option<PathBuf> {
let raw = path.to_string_lossy();
let s = raw
.strip_prefix(r"\\?\")
.or_else(|| raw.strip_prefix(r"//?/"))
.unwrap_or(raw.as_ref());
let s = s
.strip_prefix(r"UNC\")
.or_else(|| s.strip_prefix(r"unc\"))
.or_else(|| s.strip_prefix(r"UNC/"))
.unwrap_or(s);
let s = s.trim_start_matches(['\\', '/']);
let is_sep = |c: u8| c == b'\\' || c == b'/';
let host_end = s.as_bytes().iter().position(|&c| is_sep(c))?;
let host = &s[..host_end];
let after_host = &s[host_end + 1..];
let share_end = after_host
.as_bytes()
.iter()
.position(|&c| is_sep(c))
.unwrap_or(after_host.len());
let share = &after_host[..share_end];
let rest = if share_end < after_host.len() {
&after_host[share_end + 1..]
} else {
""
};
if share.len() != 2 {
return None;
}
let drive = share.as_bytes()[0];
if !drive.is_ascii_alphabetic() || share.as_bytes()[1] != b'$' {
return None;
}
if !windows_unc_host_is_local(host) {
return None;
}
let letter = (drive as char).to_ascii_uppercase();
if rest.is_empty() {
Some(PathBuf::from(format!(r"{letter}:\")))
} else {
Some(PathBuf::from(format!(r"{letter}:\{rest}")))
}
}
#[cfg(windows)]
fn windows_unc_host_is_local(host: &str) -> bool {
let host = host
.strip_prefix('[')
.and_then(|h| h.strip_suffix(']'))
.unwrap_or(host);
if host.eq_ignore_ascii_case("localhost") || host == "127.0.0.1" {
return true;
}
if host
.parse::<std::net::Ipv6Addr>()
.is_ok_and(|addr| addr.is_loopback())
{
return true;
}
std::env::var("COMPUTERNAME")
.map(|name| name.eq_ignore_ascii_case(host))
.unwrap_or(false)
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum AbsolutePathPolicy {
Reject,
AllowIfContained,
AllowAdditionalRoots(Vec<PathBuf>),
}
fn system_temp_directory_roots() -> Vec<PathBuf> {
let mut roots: Vec<PathBuf> = vec![std::env::temp_dir()];
#[cfg(unix)]
{
push_unique(&mut roots, PathBuf::from("/tmp"));
push_unique(&mut roots, PathBuf::from("/var/tmp"));
if let Ok(v) = std::env::var("TMPDIR") {
push_unique(&mut roots, PathBuf::from(v));
}
}
#[cfg(windows)]
{
for var in ["TEMP", "TMP"] {
if let Ok(v) = std::env::var(var) {
push_unique(&mut roots, PathBuf::from(v));
}
}
}
roots
}
fn push_unique(roots: &mut Vec<PathBuf>, p: PathBuf) {
if !roots.contains(&p) {
roots.push(p);
}
}
impl AbsolutePathPolicy {
pub fn allow_workspace_and_temp_dir() -> Self {
AbsolutePathPolicy::AllowAdditionalRoots(system_temp_directory_roots())
}
pub fn allow_additional_roots(roots: impl IntoIterator<Item = PathBuf>) -> Self {
AbsolutePathPolicy::AllowAdditionalRoots(roots.into_iter().collect())
}
}
#[derive(Debug)]
#[non_exhaustive]
pub enum ContainmentError {
AbsolutePath(String),
EmptyPath,
Escaped {
path: String,
root: String,
},
Canonicalize {
path: String,
source: std::io::Error,
},
}
impl std::fmt::Display for ContainmentError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
ContainmentError::AbsolutePath(p) => write!(f, "absolute paths are not allowed: {p}"),
ContainmentError::EmptyPath => write!(f, "path must not be empty"),
ContainmentError::Escaped { path, root } => {
write!(
f,
"path escapes workspace directory: {path} (workspace: {root})"
)
}
ContainmentError::Canonicalize { path, source } => {
write!(f, "failed to canonicalize path: {path}: {source}")
}
}
}
}
impl std::error::Error for ContainmentError {
fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
match self {
ContainmentError::Canonicalize { source, .. } => Some(source),
_ => None,
}
}
}
#[derive(Debug, Clone)]
pub struct PathGuard {
root: PathBuf,
canon_root: PathBuf,
absolute_policy: AbsolutePathPolicy,
}
impl PathGuard {
pub fn new(
root: PathBuf,
absolute_policy: AbsolutePathPolicy,
) -> Result<Self, ContainmentError> {
Self::new_with_policy(root, absolute_policy)
}
pub fn check_path(&self, path: &str) -> Result<PathBuf, ContainmentError> {
if is_blank_path(path) {
return Err(ContainmentError::EmptyPath);
}
let p = Path::new(path);
if p.is_absolute() {
let roots = self.absolute_allowed_roots(path)?;
return self.check_resolved_absolute(path, p, &roots);
}
validate_relative_depth(path, p, &self.root)?;
self.check_resolved_relative(path)
}
pub fn check_path_entry(&self, path: &str) -> Result<PathBuf, ContainmentError> {
if is_blank_path(path) {
return Err(ContainmentError::EmptyPath);
}
let p = Path::new(path);
if p.is_absolute() {
let roots = self.absolute_allowed_roots(path)?;
return self.check_entry_under_roots(path, p, &roots);
}
validate_relative_depth(path, p, &self.root)?;
let joined = self.root.join(path);
self.check_entry_under_roots(path, &joined, std::slice::from_ref(&self.canon_root))
}
pub fn root(&self) -> &Path {
&self.root
}
pub fn canon_root(&self) -> &Path {
&self.canon_root
}
pub fn would_allow(&self, path: &str) -> bool {
self.check_path(path).is_ok()
}
pub fn would_allow_entry(&self, path: &str) -> bool {
self.check_path_entry(path).is_ok()
}
fn absolute_allowed_roots(&self, path: &str) -> Result<Vec<PathBuf>, ContainmentError> {
match &self.absolute_policy {
AbsolutePathPolicy::Reject => Err(ContainmentError::AbsolutePath(path.to_string())),
AbsolutePathPolicy::AllowIfContained => Ok(vec![self.canon_root.clone()]),
AbsolutePathPolicy::AllowAdditionalRoots(extra) => {
let mut allowed = vec![self.canon_root.clone()];
for r in extra {
if let Ok(c) = safe_canonicalize(r) {
allowed.push(c);
}
}
Ok(allowed)
}
}
}
fn check_entry_under_roots(
&self,
display: &str,
path: &Path,
allowed_roots: &[PathBuf],
) -> Result<PathBuf, ContainmentError> {
let normalized = normalize_lexical(path);
let Some(file_name) = normalized.file_name() else {
return self.check_resolved_absolute(display, path, allowed_roots);
};
let parent = normalized.parent().unwrap_or_else(|| Path::new("."));
let parent_canon =
canonicalize_or_ancestor(parent).map_err(|e| ContainmentError::Canonicalize {
path: display.to_string(),
source: e,
})?;
let contained = allowed_roots
.iter()
.any(|r| path_is_under(parent_canon.as_path(), r));
if !contained {
return Err(ContainmentError::Escaped {
path: dunce::simplified(Path::new(display))
.to_string_lossy()
.into_owned(),
root: self.root.display().to_string(),
});
}
let mut entry = prefer_local_drive_path(parent_canon);
entry.push(file_name);
Ok(entry)
}
fn check_resolved_absolute(
&self,
path: &str,
p: &Path,
allowed_roots: &[PathBuf],
) -> Result<PathBuf, ContainmentError> {
let canon = canonicalize_or_ancestor(p).map_err(|e| ContainmentError::Canonicalize {
path: path.to_string(),
source: e,
})?;
let contained = allowed_roots.iter().any(|r| path_is_under(&canon, r));
if !contained {
return Err(ContainmentError::Escaped {
path: dunce::simplified(Path::new(path))
.to_string_lossy()
.into_owned(),
root: self.root.display().to_string(),
});
}
Ok(prefer_local_drive_path(canon))
}
fn check_resolved_relative(&self, path: &str) -> Result<PathBuf, ContainmentError> {
let joined = self.root.join(path);
let canon =
canonicalize_or_ancestor(&joined).map_err(|e| ContainmentError::Canonicalize {
path: path.to_string(),
source: e,
})?;
if !path_is_under(&canon, &self.canon_root) {
return Err(ContainmentError::Escaped {
path: dunce::simplified(Path::new(path))
.to_string_lossy()
.into_owned(),
root: self.root.display().to_string(),
});
}
Ok(prefer_local_drive_path(canon))
}
}
pub struct PathGuardBuilder {
root: PathBuf,
policy: AbsolutePathPolicy,
}
impl PathGuard {
pub fn builder(root: PathBuf) -> PathGuardBuilder {
PathGuardBuilder {
root,
policy: AbsolutePathPolicy::Reject,
}
}
pub fn new_with_policy(
root: PathBuf,
policy: AbsolutePathPolicy,
) -> Result<Self, ContainmentError> {
let canon_root = safe_canonicalize(&root).map_err(|e| ContainmentError::Canonicalize {
path: root.display().to_string(),
source: e,
})?;
Ok(Self {
root,
canon_root,
absolute_policy: policy,
})
}
}
impl PathGuardBuilder {
pub fn allow_temp_directory(mut self) -> Self {
let temps = system_temp_directory_roots();
self.policy = match self.policy {
AbsolutePathPolicy::Reject | AbsolutePathPolicy::AllowIfContained => {
AbsolutePathPolicy::AllowAdditionalRoots(temps)
}
AbsolutePathPolicy::AllowAdditionalRoots(mut roots) => {
for t in temps {
push_unique(&mut roots, t);
}
AbsolutePathPolicy::AllowAdditionalRoots(roots)
}
};
self
}
pub fn allow_root(mut self, additional: impl Into<PathBuf>) -> Self {
let extra = additional.into();
self.policy = match self.policy {
AbsolutePathPolicy::Reject | AbsolutePathPolicy::AllowIfContained => {
AbsolutePathPolicy::AllowAdditionalRoots(vec![extra])
}
AbsolutePathPolicy::AllowAdditionalRoots(mut roots) => {
push_unique(&mut roots, extra);
AbsolutePathPolicy::AllowAdditionalRoots(roots)
}
};
self
}
pub fn build(self) -> Result<PathGuard, ContainmentError> {
PathGuard::new_with_policy(self.root, self.policy)
}
}
fn validate_relative_depth(path: &str, p: &Path, root: &Path) -> Result<(), ContainmentError> {
let mut depth: i32 = 0;
for component in p.components() {
match component {
Component::ParentDir => {
depth -= 1;
if depth < 0 {
return Err(ContainmentError::Escaped {
path: path.to_string(),
root: root.display().to_string(),
});
}
}
Component::Normal(_) => {
depth += 1;
}
Component::CurDir => {}
_ => {
return Err(ContainmentError::Escaped {
path: path.to_string(),
root: root.display().to_string(),
});
}
}
}
Ok(())
}
fn normalize_lexical(path: &Path) -> PathBuf {
use std::path::Component;
let mut parts: Vec<Component<'_>> = Vec::new();
for c in path.components() {
match c {
Component::ParentDir => {
if matches!(parts.last(), Some(Component::Normal(_))) {
parts.pop();
} else {
parts.push(c);
}
}
Component::CurDir => { }
_ => parts.push(c),
}
}
parts.iter().collect()
}
fn canonicalize_or_ancestor(path: &Path) -> std::io::Result<PathBuf> {
let openable = prefer_openable_path(path);
let normalized = normalize_lexical(&openable);
let path = normalized.as_path();
if path.exists() {
return safe_canonicalize(path);
}
let mut ancestor = path;
let mut tail_components = Vec::new();
loop {
match ancestor.parent() {
Some(p) if p.exists() => {
if let Some(file_name) = ancestor.file_name() {
tail_components.push(file_name.to_os_string());
}
let canon_ancestor = safe_canonicalize(p)?;
let mut result = canon_ancestor;
for c in tail_components.into_iter().rev() {
result.push(c);
}
return Ok(result);
}
Some(p) => {
if let Some(file_name) = ancestor.file_name() {
tail_components.push(file_name.to_os_string());
}
ancestor = p;
}
None => return safe_canonicalize(path),
}
}
}
const _: () = {
fn _assert<T: Send + Sync>() {}
let _ = _assert::<PathGuard>;
let _ = _assert::<AbsolutePathPolicy>;
let _ = _assert::<ContainmentError>;
};
#[cfg(test)]
#[path = "containment_tests.rs"]
mod tests;