use std::borrow::Cow;
use super::regions::classify_region;
use crate::engine::facet::LocalLocus;
pub(crate) fn read_locus(path: &str) -> LocalLocus {
face(path, false)
}
pub(crate) fn write_locus(path: &str) -> LocalLocus {
face(path, true)
}
fn face(path: &str, want_write: bool) -> LocalLocus {
if let Some(local) = file_url_local(path) {
return classify_local(local, want_write);
}
if is_network_url(path) {
if path.contains('$') || path.contains("__SAFE_CHAINS_CMDSUB__") || url_escapes_cwd(path) {
return LocalLocus::Machine;
}
return LocalLocus::Worktree;
}
classify_local(path, want_write)
}
fn url_escapes_cwd(url: &str) -> bool {
let mut depth: i32 = 0;
for seg in url.split('/') {
match seg {
"" | "." => {}
".." => {
depth -= 1;
if depth < 0 {
return true;
}
}
_ => depth += 1,
}
}
false
}
const IFS_WHITESPACE: [char; 3] = [' ', '\t', '\n'];
fn classify_local(path: &str, want_write: bool) -> LocalLocus {
let whole = classify_one(path, want_write);
let expanded = crate::pathctx::expand_vars(path, want_write);
if !expanded.contains(IFS_WHITESPACE) {
return whole;
}
expanded
.split(IFS_WHITESPACE)
.filter(|piece| !piece.is_empty())
.map(|piece| classify_one(piece, want_write))
.fold(whole, LocalLocus::max)
}
fn classify_one(path: &str, want_write: bool) -> LocalLocus {
let expanded = crate::pathctx::expand_vars(path, want_write);
let expanded = Cow::Owned(neutralize_atoms(&expanded).into_owned());
let (operand_tag, base) = match tagged_substitution(&expanded) {
Some((tag, rewritten)) => (Some(tag), Cow::Owned(rewritten)),
None => (None, expanded),
};
let resolved = crate::pathctx::resolve(&base);
let (cwd_tag, base) = match tagged_substitution(&resolved) {
Some((tag, rewritten)) => (Some(tag), Cow::Owned(rewritten)),
None => (None, resolved),
};
let plain = classify_pinned(&base, want_write);
[operand_tag, cwd_tag].into_iter().flatten().fold(plain, LocalLocus::max)
}
fn classify_pinned(resolved: &str, want_write: bool) -> LocalLocus {
let canonical = canonicalize(resolved);
if is_unpinnable(&canonical) {
return LocalLocus::Machine;
}
let role = classify_region(&canonical);
if want_write { role.write_locus } else { role.read_locus }
}
const SUB_STANDIN: &str = "sc_substitution_value";
fn canonicalize(path: &str) -> Cow<'_, str> {
let home = std::env::var("HOME").ok();
let home_abs = home
.as_deref()
.filter(|h| !h.is_empty() && path.strip_prefix(*h).is_some_and(|r| r.is_empty() || r.starts_with('/')));
let dotty = path.contains("//") || path.contains("/./") || path.ends_with("/.");
if home_abs.is_none() && !dotty {
return Cow::Borrowed(path);
}
let tilded = match home_abs {
Some(h) if path.len() == h.len() => "~".to_string(),
Some(h) => format!("~{}", &path[h.len()..]),
None => path.to_string(),
};
if !(tilded.contains("//") || tilded.contains("/./") || tilded.ends_with("/.")) {
return Cow::Owned(tilded);
}
let absolute = tilded.starts_with('/');
let joined = tilded
.split('/')
.filter(|seg| !seg.is_empty() && *seg != ".")
.collect::<Vec<_>>()
.join("/");
Cow::Owned(if absolute { format!("/{joined}") } else { joined })
}
fn file_url_local(path: &str) -> Option<&str> {
if path.len() < 5 || !path.as_bytes()[..5].eq_ignore_ascii_case(b"file:") {
return None;
}
let rest = &path[5..];
Some(rest.strip_prefix("//").map_or(rest, |authority| {
authority.find('/').map_or("", |i| &authority[i..])
}))
}
fn is_network_url(path: &str) -> bool {
let Some(idx) = path.find("://") else {
return false;
};
let scheme = &path[..idx];
scheme.starts_with(|c: char| c.is_ascii_alphabetic())
&& scheme.bytes().all(|b| b.is_ascii_alphanumeric() || matches!(b, b'+' | b'-' | b'.'))
}
pub(crate) fn classify_locus(path: &str) -> LocalLocus {
write_locus(path)
}
pub(crate) fn is_unpinnable(path: &str) -> bool {
path.contains('$')
|| path.contains("__SAFE_CHAINS_CMDSUB__")
|| path.contains(crate::cst::eval::ATOM_SENTINEL)
|| is_parent_escape(path)
}
fn tagged_substitution(path: &str) -> Option<(LocalLocus, String)> {
use crate::engine::facet::FacetTerm;
let prefix_len = crate::cst::eval::TAGGED_PREFIX.len();
let mut worst: Option<LocalLocus> = None;
let mut out = String::with_capacity(path.len());
let mut rest = path;
while let Some(at) = rest.find(crate::cst::eval::TAGGED_PREFIX) {
let after = &rest[at + prefix_len..];
let Some((term, tail)) = after.split_once("__") else {
return Some((LocalLocus::Machine, String::new()));
};
let Some(locus) = LocalLocus::from_term(&term.to_lowercase().replace('_', "-")) else {
return Some((LocalLocus::Machine, String::new()));
};
out.push_str(&rest[..at]);
out.push_str(SUB_STANDIN);
worst = Some(worst.map_or(locus, |w: LocalLocus| w.max(locus)));
rest = tail;
}
let worst = worst?;
out.push_str(rest);
if out.contains('$') || out.contains(UNPINNABLE_MARK) {
return Some((LocalLocus::Machine, String::new()));
}
Some((worst, out))
}
const UNPINNABLE_MARK: &str = "__SAFE_CHAINS_CMDSUB__";
fn neutralize_atoms(path: &str) -> Cow<'_, str> {
let atom = crate::cst::eval::ATOM_SENTINEL;
if !path.contains(atom) {
return Cow::Borrowed(path);
}
let out = path
.split('/')
.map(|comp| {
if !comp.contains(atom) {
return comp.to_string();
}
let residue = comp.replace(atom, "");
if residue.is_empty()
|| residue.chars().all(|c| c == '.')
|| residue.contains(crate::cst::eval::TAGGED_PREFIX)
{
return UNPINNABLE_MARK.to_string();
}
comp.replace(atom, SUB_STANDIN)
})
.collect::<Vec<_>>()
.join("/");
Cow::Owned(out)
}
pub(crate) fn is_substitution_value(path: &str) -> bool {
path.contains(crate::cst::eval::TAGGED_PREFIX)
}
#[cfg(test)]
mod atom_backstop {
#[test]
fn an_atom_sentinel_is_unpinnable_without_neutralization() {
let atom = crate::cst::eval::ATOM_SENTINEL;
assert!(super::is_unpinnable(atom), "a bare atom sentinel must fail closed");
assert!(
super::is_unpinnable(&format!("~/.ssh/dx_{atom}.txt")),
"an un-neutralized atom must fail closed wherever it appears"
);
assert!(
!super::is_unpinnable(&format!("~/.ssh/dx_{}.txt", super::SUB_STANDIN)),
"the NEUTRALIZED form must be pinnable, or confinement could never pay off"
);
}
}
pub(crate) fn names_credential_store(path: &str) -> bool {
let expanded = crate::pathctx::expand_vars(path, false);
let neutralized = neutralize_atoms(&expanded);
classify_region(&crate::pathctx::resolve(&neutralized)).reads_secret
}
pub(crate) fn anchoring_of(path: &str) -> crate::engine::facet::Anchoring {
use crate::engine::facet::Anchoring;
let neutralized = neutralize_atoms(path);
if is_unpinnable(&neutralized) {
Anchoring::Opaque
} else if neutralized.as_ref() != path {
Anchoring::Anchored
} else {
Anchoring::Literal
}
}
fn is_parent_escape(path: &str) -> bool {
path == ".." || path.starts_with("../") || path.contains("/../") || path.ends_with("/..")
}
#[cfg(test)]
mod tests {
use super::*;
use crate::engine::facet::LocalLocus;
#[test]
fn the_unpinnable_guard_worst_cases_both_faces() {
for p in ["$HOME/.ssh/id_rsa", "$OUT/file", "../secret", "a/../../etc/passwd", "dir/.."] {
assert_eq!(read_locus(p), LocalLocus::Machine, "read {p}");
assert_eq!(write_locus(p), LocalLocus::Machine, "write {p}");
}
}
#[test]
fn write_face_admits_only_workspace_temp_and_streams() {
assert_eq!(write_locus("/dev/null"), LocalLocus::Process);
assert_eq!(write_locus("/tmp/scratch"), LocalLocus::Temp);
assert_eq!(write_locus("notes.md"), LocalLocus::Worktree);
assert_eq!(write_locus("src/engine/mod.rs"), LocalLocus::Worktree);
assert_eq!(write_locus(".git/config"), LocalLocus::WorktreeTrusted, "in-project but write-frozen");
assert_eq!(write_locus("~"), LocalLocus::Machine);
assert_eq!(write_locus("~/notes"), LocalLocus::Machine);
assert_eq!(write_locus("/etc/hosts"), LocalLocus::Machine);
assert_eq!(write_locus("/usr/local/bin/x"), LocalLocus::Machine);
assert_eq!(write_locus("/dev/rdisk0"), LocalLocus::Machine);
assert_eq!(write_locus("~bob/.ssh/id_rsa"), LocalLocus::Machine, "another user's home");
}
#[test]
fn read_face_admits_only_the_workspace_not_system_paths() {
assert_eq!(read_locus("/etc/hosts"), LocalLocus::Machine);
assert_eq!(read_locus("/usr/bin/python3"), LocalLocus::Machine);
assert_eq!(read_locus("/etc/shadow"), LocalLocus::Machine);
assert_eq!(read_locus("~/.ssh/id_rsa"), LocalLocus::Machine);
assert_eq!(read_locus("~/notes"), LocalLocus::Machine);
assert_eq!(read_locus("/some/unmapped/thing"), LocalLocus::Machine);
assert_eq!(read_locus("notes.md"), LocalLocus::Worktree);
assert_eq!(read_locus("/tmp/x"), LocalLocus::Temp);
}
#[test]
fn file_urls_classify_the_local_path_they_name() {
for p in [
"file:///etc/shadow",
"file://localhost/etc/shadow",
"file:/etc/shadow",
"FILE:///etc/shadow",
"File:///etc/shadow",
] {
assert_eq!(read_locus(p), LocalLocus::Machine, "read {p}");
assert_eq!(write_locus(p), LocalLocus::Machine, "write {p}");
}
assert_eq!(read_locus("file:///etc/hosts"), LocalLocus::Machine);
assert_eq!(write_locus("file:///etc/hosts"), LocalLocus::Machine);
assert_eq!(read_locus("file:notes.txt"), LocalLocus::Worktree);
assert_eq!(read_locus("file://../../etc/shadow"), LocalLocus::Machine);
}
#[test]
fn network_urls_are_not_local_operations() {
for p in ["http://example.com/a", "https://x/a/../b", "ftp://h/f", "s3://bucket/key", "ssh://h/p"] {
assert_eq!(read_locus(p), LocalLocus::Worktree, "read {p}");
assert_eq!(write_locus(p), LocalLocus::Worktree, "write {p}");
}
assert_eq!(write_locus("/tmp/weird://name"), LocalLocus::Temp);
assert_eq!(read_locus("./a:b"), LocalLocus::Worktree);
}
#[test]
fn a_scheme_url_that_net_escapes_cwd_is_not_admitted() {
assert_eq!(read_locus("https://x/a/../b"), LocalLocus::Worktree);
assert_eq!(read_locus("s3://bucket/../key"), LocalLocus::Worktree);
for p in ["s3://../../secret.txt", "gopher://../../etc/passwd", "s3://a/../../../etc/x"] {
assert_eq!(read_locus(p), LocalLocus::Machine, "read {p}");
assert_eq!(write_locus(p), LocalLocus::Machine, "write {p}");
}
assert_eq!(read_locus("s3://$SECRET/x"), LocalLocus::Machine);
}
#[test]
fn canonicalize_folds_equivalent_spellings() {
assert_eq!(canonicalize("~/.config//safe-chains.toml"), "~/.config/safe-chains.toml");
assert_eq!(canonicalize("~/.config/./safe-chains.toml"), "~/.config/safe-chains.toml");
assert_eq!(canonicalize("/a//b/./c"), "/a/b/c");
assert_eq!(canonicalize("~/a/../b"), "~/a/../b");
assert_eq!(canonicalize("~/.config/safe-chains.toml"), "~/.config/safe-chains.toml");
if let Some(home) = std::env::var("HOME").ok().filter(|h| h.starts_with('/')) {
assert_eq!(canonicalize(&format!("{home}/.config/safe-chains.toml")), "~/.config/safe-chains.toml");
}
}
#[test]
fn credential_stores_are_named_as_such() {
assert!(names_credential_store("~/.ssh/id_rsa"));
assert!(names_credential_store("~/.aws/credentials"));
assert!(names_credential_store("~/.gnupg/secring.gpg"));
assert!(!names_credential_store("/etc/hosts")); assert!(!names_credential_store("notes.md"));
assert!(names_credential_store("~/.ssh/__SAFE_CHAINS_CMDSUB__"));
assert!(names_credential_store(&format!("~/.ssh/dx_{}.txt", crate::cst::eval::ATOM_SENTINEL)));
assert!(!names_credential_store("~/projects/other/__SAFE_CHAINS_CMDSUB__"));
}
use proptest::prelude::*;
proptest! {
#[test]
fn a_dollar_anywhere_forces_machine(s in ".{0,30}") {
prop_assert_eq!(read_locus(&format!("{s}$")), LocalLocus::Machine);
prop_assert_eq!(write_locus(&format!("{s}$")), LocalLocus::Machine);
}
#[test]
fn a_parent_escape_forces_machine(s in "[a-zA-Z0-9/_]{0,20}") {
prop_assert_eq!(write_locus(&format!("{s}/../x")), LocalLocus::Machine);
prop_assert_eq!(write_locus(&format!("../{s}")), LocalLocus::Machine);
}
}
}