//! 工作区路径的语义化规范化（`..` / `.`）与根边界校验（**工具与 Web 共用单一真源**）。
//!
//! ## 边界说明
//!
//! - **前缀校验**使用 [`Path::starts_with`]（按路径分量），避免 `/foo/bar` 误匹配 `/foo/bar-baz`。
//! - **`..` / 相对路径**：由 [`path_absolutize::Absolutize`] 在已 canonical 的根下解析；对**已存在路径**再 `canonicalize` 以解析符号链接真实位置。
//!
//! ## 校验与打开之间的竞态（TOCTOU）
//!
//! 在**信任工作区**的典型开发场景下，本模块在访问前对路径做 `canonicalize` 与 `starts_with` 检查，可拒绝在**检查时刻**已指向根外的符号链接等情形。
//!
//! **已落地缓解**（见 [`crate::cm_tools::workspace::fs`]）：在 **Unix** 上，`read_file` 等经 **`resolve_for_read_open`** 在已打开的工作区根 fd 上使用 **`openat2` + `RESOLVE_IN_ROOT`（Linux）** 或单次 `File::open`（其它 Unix）打开目标，将「策略校验 ↔ 业务打开」之间的窗口收窄；**Web** 工作区列表/读文件/写文件/删文件在 Unix 上同样经该模块做目录或文件打开。工作区内 symlink 仍可被跟随，但解析不得越过该根。
//!
//! **残余风险**：策略校验仍依赖校验时刻的 `canonicalize`；**非 Linux** 或未走 [`crate::cm_tools::workspace::fs`] 的路径仍可能存在竞态；目录删除等操作未完全 `openat` 化。**不要**将当前实现等同于内核级「不可逃逸」保证；多租户或不可信工作区须与 **HTTP 鉴权**等一并评估。
//!
//! 用户可见说明见 **`README.md`**、**`docs/配置说明.md`**（工作区）。工具侧解析与打开入口见 **`src/tools/file/path.rs`**。
use path_absolutize::Absolutize;
use std::path::{Path, PathBuf};
use thiserror::Error;
use crate::cm_config::AgentConfig;
/// Web 与工作区根组合的相对 `path` 参数允许的最大字节数（原始字符串，trim 前测量）。
pub const WEB_WORKSPACE_REL_SUBPATH_MAX_BYTES: usize = 8192;
/// 工作区路径解析与策略校验失败（可判别类别，供日志与调用方分支；对用户展示用 [`Display`] / [`WorkspacePathError::user_message`]）。
#[derive(Debug, Error)]
pub enum WorkspacePathError {
/// `path` 等入参为空或仅空白。
#[error("path 不能为空")]
EmptyPath,
/// 要求相对路径时收到了绝对路径。
#[error("路径必须为相对于工作目录的相对路径，不能使用绝对路径")]
AbsolutePathNotAllowed,
/// 切换工作区时路径参数为空。
#[error("路径不能为空")]
WorkspaceSetPathEmpty,
/// 无法取得当前工作目录（如 `getcwd` 失败）。
#[error("无法获取当前目录: {0}")]
CurrentDirUnavailable(#[source] std::io::Error),
/// 路径不存在、无法 canonicalize，或不是目录等与解析/存在性相关的问题。
#[error("工作区路径无效或不存在: {0}")]
WorkspacePathInvalid(#[source] std::io::Error),
/// 通用「无法解析 canonical 路径」（工具读文件、祖先校验等）。
#[error("路径无法解析: {0}")]
PathResolveFailed(#[source] std::io::Error),
/// 工作区根或当前目录无法 canonicalize。
#[error("工作目录无法解析: {0}")]
WorkspaceResolveFailed(#[source] std::io::Error),
/// Web 尚未通过 `POST /workspace` 选择工作区（`effective_workspace_path` 为空，**不要**对空路径 `canonicalize`）。
#[error("请先设置工作区")]
WebEffectiveWorkspaceUnset,
/// 未配置 `web_workspace_pool`，无法按项目名切换工作区。
#[error("未配置 Web 工作区项目池（web_workspace_pool）")]
WebWorkspacePoolDisabled,
/// 项目名不合法。
#[error("项目名无效: {0}")]
InvalidProjectName(String),
/// 路径存在但不是目录。
#[error("工作区路径必须是已存在的目录")]
NotADirectory,
/// 命中敏感系统目录前缀黑名单。
#[error("工作区路径命中敏感目录黑名单，请选择业务目录")]
SensitivePathDenied,
/// 工作区根命中敏感前缀（与切换路径文案略异，便于区分场景）。
#[error("工作区根路径命中敏感目录黑名单")]
EffectiveRootSensitive,
/// 路径落在允许根集合之外（策略拒绝）。
#[error("工作区路径不在允许范围内（须位于以下根目录之一下: {roots_display})")]
OutsideAllowedRoots { roots_display: String },
/// 当前生效工作区根不在允许范围内。
#[error("工作区根不在允许范围内（须位于以下根目录之一: {roots_display})")]
EffectiveRootOutsideAllowed { roots_display: String },
/// 规范化后路径越过工作区根（`..` 逃逸或 Web 子路径越界）。
#[error("路径不能超出工作目录")]
OutsideWorkspaceRoot,
/// Web 查询参数中的相对路径字节过长（参见 [`WEB_WORKSPACE_REL_SUBPATH_MAX_BYTES`]）。
#[error("path 过长（上限 {max} 字节）")]
WebRelSubpathTooLong { max: usize },
/// `path_absolutize` 词法规范化失败（`absolutize` / `absolutize_from` 的 IO 错误）。
#[error("路径规范化失败: {0}")]
NormalizationFailed(#[source] std::io::Error),
/// 自根向上找不到任何存在祖先（极少见，如根被删）。
#[error("路径无法解析")]
NoExistingAncestor,
}
impl WorkspacePathError {
/// 与历史 `String` 错误语义一致的简短分类，便于 metrics / 结构化日志（不含敏感路径全量时可只记此项）。
#[must_use]
pub fn kind(&self) -> &'static str {
match self {
WorkspacePathError::EmptyPath => "empty_path",
WorkspacePathError::AbsolutePathNotAllowed => "absolute_path_not_allowed",
WorkspacePathError::WorkspaceSetPathEmpty => "workspace_set_path_empty",
WorkspacePathError::CurrentDirUnavailable(_) => "current_dir_unavailable",
WorkspacePathError::WorkspacePathInvalid(_) => "workspace_path_invalid",
WorkspacePathError::PathResolveFailed(_) => "path_resolve_failed",
WorkspacePathError::WorkspaceResolveFailed(_) => "workspace_resolve_failed",
WorkspacePathError::WebEffectiveWorkspaceUnset => "web_effective_workspace_unset",
WorkspacePathError::WebWorkspacePoolDisabled => "web_workspace_pool_disabled",
WorkspacePathError::InvalidProjectName(_) => "invalid_project_name",
WorkspacePathError::NotADirectory => "not_a_directory",
WorkspacePathError::SensitivePathDenied => "sensitive_path_denied",
WorkspacePathError::EffectiveRootSensitive => "effective_root_sensitive",
WorkspacePathError::OutsideAllowedRoots { .. } => "outside_allowed_roots",
WorkspacePathError::EffectiveRootOutsideAllowed { .. } => {
"effective_root_outside_allowed"
}
WorkspacePathError::OutsideWorkspaceRoot => "outside_workspace_root",
WorkspacePathError::WebRelSubpathTooLong { .. } => "web_rel_subpath_too_long",
WorkspacePathError::NormalizationFailed(_) => "path_normalize_failed",
WorkspacePathError::NoExistingAncestor => "no_existing_ancestor",
}
}
/// 是否属于「策略/权限」类（越界、敏感目录、允许根外）；用于 HTTP 403 等映射。
#[must_use]
pub fn is_policy_denied(&self) -> bool {
matches!(
self,
WorkspacePathError::SensitivePathDenied
| WorkspacePathError::EffectiveRootSensitive
| WorkspacePathError::OutsideAllowedRoots { .. }
| WorkspacePathError::EffectiveRootOutsideAllowed { .. }
| WorkspacePathError::OutsideWorkspaceRoot
| WorkspacePathError::AbsolutePathNotAllowed
)
}
/// 面向用户/API 的说明（与实现 `Display` 一致，便于显式调用）。
#[must_use]
pub fn user_message(&self) -> String {
self.to_string()
}
}
/// 校验用于切换工作区根的 `path`（Web **`POST /workspace`** 与 REPL **`/workspace`** 共用）。
///
/// 须为已存在目录，`canonicalize` 后落在 **`workspace_allowed_roots`** 内且不得命中敏感路径黑名单。
/// 相对路径相对于**进程当前工作目录**解析（与历史 Web 行为一致）。
pub fn validate_workspace_set_path(
cfg: &AgentConfig,
raw: &str,
) -> Result<PathBuf, WorkspacePathError> {
let raw = raw.trim();
if raw.is_empty() {
return Err(WorkspacePathError::WorkspaceSetPathEmpty);
}
let cwd = std::env::current_dir().map_err(WorkspacePathError::CurrentDirUnavailable)?;
let p = Path::new(raw);
let joined = if p.is_absolute() {
p.to_path_buf()
} else {
cwd.join(p)
};
let canon = joined
.canonicalize()
.map_err(WorkspacePathError::WorkspacePathInvalid)?;
if !canon.is_dir() {
return Err(WorkspacePathError::NotADirectory);
}
if is_sensitive_workspace_path(&canon) {
return Err(WorkspacePathError::SensitivePathDenied);
}
if !is_within_allowed_roots(&canon, &cfg.workspace_roots.workspace_allowed_roots) {
let roots_display = cfg
.workspace_roots
.workspace_allowed_roots
.iter()
.map(|p| p.display().to_string())
.collect::<Vec<_>>()
.join(", ");
return Err(WorkspacePathError::OutsideAllowedRoots { roots_display });
}
Ok(canon)
}
/// 将项目池中的项目名解析为绝对路径并校验（目录须已存在，规则同 [`validate_workspace_set_path`]）。
pub fn validate_workspace_project_set_path(
cfg: &AgentConfig,
raw_name: &str,
) -> Result<PathBuf, WorkspacePathError> {
let pool = cfg
.workspace_roots
.web_workspace_pool
.as_ref()
.ok_or(WorkspacePathError::WebWorkspacePoolDisabled)?;
let dir = super::project::workspace_project_dir(pool.as_path(), raw_name)
.map_err(|e| WorkspacePathError::InvalidProjectName(e.to_string()))?;
validate_workspace_set_path(cfg, &dir.display().to_string())
}
/// Web `POST /workspace` 与「当前会话工作区根」校验共用的敏感路径前缀（canonical 后命中即拒绝）。
const SENSITIVE_WORKSPACE_PREFIXES: &[&str] = &[
"/proc", "/sys", "/dev", "/etc", "/boot", "/root", "/bin", "/sbin", "/usr",
];
/// 将工作目录（可为符号链接）解析为 **canonical** 绝对路径，供工具与 Web 共用。
pub fn canonical_workspace_root(base: &Path) -> Result<PathBuf, WorkspacePathError> {
base.canonicalize()
.map_err(WorkspacePathError::WorkspaceResolveFailed)
}
/// 规范化后的路径是否命中敏感系统目录前缀（用于拒绝把工作区设到或解析到此类路径）。
pub fn is_sensitive_workspace_path(path: &Path) -> bool {
SENSITIVE_WORKSPACE_PREFIXES.iter().any(|prefix| {
let p = Path::new(prefix);
path == p || path.starts_with(p)
})
}
/// `candidate` 与 `root` 均须已为 **canonical** 路径；要求 `candidate == root` 或 `candidate` 为 `root` 之下的子孙路径。错误文案与工具层越界一致。
pub fn ensure_canonical_within_root(
candidate: &Path,
root: &Path,
) -> Result<(), WorkspacePathError> {
if candidate.starts_with(root) {
Ok(())
} else {
Err(WorkspacePathError::OutsideWorkspaceRoot)
}
}
/// `candidate`（已 canonical）是否落在任一 **canonical** 允许根之下（配置 `workspace_allowed_roots`）。
/// 当 `roots` 为空时，允许任意路径（返回 true）。
pub fn is_within_allowed_roots(candidate: &Path, roots: &[PathBuf]) -> bool {
if roots.is_empty() {
return true;
}
roots.iter().any(|r| candidate.starts_with(r))
}
/// 校验「当前生效的工作区根」仍合法：非敏感目录且在 `workspace_allowed_roots` 内。
pub fn validate_effective_workspace_base(
cfg: &AgentConfig,
base_canonical: &Path,
) -> Result<(), WorkspacePathError> {
if is_sensitive_workspace_path(base_canonical) {
return Err(WorkspacePathError::EffectiveRootSensitive);
}
if !is_within_allowed_roots(base_canonical, &cfg.workspace_roots.workspace_allowed_roots) {
let roots_display = cfg
.workspace_roots
.workspace_allowed_roots
.iter()
.map(|p| p.display().to_string())
.collect::<Vec<_>>()
.join(", ");
return Err(WorkspacePathError::EffectiveRootOutsideAllowed { roots_display });
}
Ok(())
}
/// 自 `target` 向上找到最近存在路径并 canonicalize，须落在 `root_canonical` 下（写入路径防 symlink 逃逸）。
pub fn ensure_existing_ancestor_within_root(
root_canonical: &Path,
target: &Path,
) -> Result<(), WorkspacePathError> {
let mut ancestor = target;
while !ancestor.exists() {
ancestor = ancestor
.parent()
.ok_or(WorkspacePathError::NoExistingAncestor)?;
}
let ancestor_canonical = ancestor
.canonicalize()
.map_err(WorkspacePathError::PathResolveFailed)?;
ensure_canonical_within_root(&ancestor_canonical, root_canonical)
}
/// `sub` 必须为相对路径；在已 canonical 的 `workspace_root` 下解析并去掉 `.` / `..`，且不得越出根。
pub fn absolutize_relative_under_root(
workspace_root: &Path,
sub: &str,
) -> Result<PathBuf, WorkspacePathError> {
let sub = sub.trim();
if sub.is_empty() {
return Err(WorkspacePathError::EmptyPath);
}
if Path::new(sub).is_absolute() {
return Err(WorkspacePathError::AbsolutePathNotAllowed);
}
let normalized = Path::new(sub).absolutize_from(workspace_root);
if !normalized.starts_with(workspace_root) {
return Err(WorkspacePathError::OutsideWorkspaceRoot);
}
Ok(normalized.into_owned())
}
/// Web 工作区写入等：`sub` 可为绝对或相对路径；规范化后须落在 `base_canonical` 之下。
pub fn absolutize_workspace_subpath(
base_canonical: &Path,
sub: &str,
) -> Result<PathBuf, WorkspacePathError> {
let sub = sub.trim();
if sub.is_empty() {
return Err(WorkspacePathError::EmptyPath);
}
let normalized = if Path::new(sub).is_absolute() {
Path::new(sub)
.absolutize()
.map_err(WorkspacePathError::NormalizationFailed)?
} else {
Path::new(sub).absolutize_from(base_canonical)
};
if !normalized.starts_with(base_canonical) {
return Err(WorkspacePathError::OutsideWorkspaceRoot);
}
Ok(normalized.into_owned())
}
/// Web：在已 canonical 的工作区根下解析只读路径；`sub` 缺省或空（trim 后）则返回根本身。
pub fn resolve_web_workspace_read_path(
base_canonical: &Path,
sub: Option<&str>,
) -> Result<PathBuf, WorkspacePathError> {
if let Some(s) = sub
&& s.len() > WEB_WORKSPACE_REL_SUBPATH_MAX_BYTES
{
return Err(WorkspacePathError::WebRelSubpathTooLong {
max: WEB_WORKSPACE_REL_SUBPATH_MAX_BYTES,
});
}
let sub_trimmed = match sub {
Some(s) if !s.trim().is_empty() => s.trim(),
_ => return Ok(base_canonical.to_path_buf()),
};
let normalized = absolutize_workspace_subpath(base_canonical, sub_trimmed)?;
let canonical = normalized
.canonicalize()
.map_err(WorkspacePathError::PathResolveFailed)?;
ensure_canonical_within_root(&canonical, base_canonical)?;
Ok(canonical)
}
/// Web：解析写入路径（目标可不存在）；防 symlink 逃逸。
pub fn resolve_web_workspace_write_path(
base_canonical: &Path,
sub: &str,
) -> Result<PathBuf, WorkspacePathError> {
let sub = sub.trim();
if sub.is_empty() {
return Err(WorkspacePathError::EmptyPath);
}
let normalized = absolutize_workspace_subpath(base_canonical, sub)?;
ensure_existing_ancestor_within_root(base_canonical, &normalized)?;
Ok(normalized)
}
#[cfg(test)]
mod tests {
use super::*;
use std::path::Path;
#[test]
fn sensitive_workspace_path_matches_prefixes() {
assert!(is_sensitive_workspace_path(Path::new("/proc")));
assert!(is_sensitive_workspace_path(Path::new("/proc/1234")));
assert!(is_sensitive_workspace_path(Path::new("/etc/nginx")));
assert!(!is_sensitive_workspace_path(Path::new("/workspace")));
assert!(!is_sensitive_workspace_path(Path::new("/tmp/project")));
}
#[test]
fn starts_with_root_is_component_wise() {
let root = Path::new("/tmp/workspace");
let inside = Path::new("/tmp/workspace/project");
let sibling = Path::new("/tmp/workspace2");
assert!(ensure_canonical_within_root(inside, root).is_ok());
assert!(ensure_canonical_within_root(sibling, root).is_err());
}
#[test]
fn validate_workspace_set_path_rejects_empty() {
let cfg = crate::cm_config::load_config(None).expect("embedded default config");
let e = validate_workspace_set_path(&cfg, " ").expect_err("empty");
assert_eq!(e.kind(), "workspace_set_path_empty");
let msg = e.user_message();
assert!(msg.contains("空"), "{msg}");
}
#[test]
fn outside_workspace_root_kind() {
let e = WorkspacePathError::OutsideWorkspaceRoot;
assert_eq!(e.kind(), "outside_workspace_root");
}
#[test]
fn resolve_web_rejects_oversized_subpath() {
use tempfile::tempdir;
let dir = tempdir().expect("tempdir");
let base = dir.path().canonicalize().expect("canon");
let long = "a".repeat(WEB_WORKSPACE_REL_SUBPATH_MAX_BYTES + 1);
let e = resolve_web_workspace_read_path(&base, Some(&long)).expect_err("oversized");
assert!(matches!(e, WorkspacePathError::WebRelSubpathTooLong { .. }));
}
#[test]
fn is_policy_denied_true_for_boundary_and_sensitive_variants() {
assert!(WorkspacePathError::SensitivePathDenied.is_policy_denied());
assert!(WorkspacePathError::EffectiveRootSensitive.is_policy_denied());
assert!(WorkspacePathError::OutsideWorkspaceRoot.is_policy_denied());
assert!(WorkspacePathError::AbsolutePathNotAllowed.is_policy_denied());
assert!(
WorkspacePathError::OutsideAllowedRoots {
roots_display: "/tmp".into(),
}
.is_policy_denied()
);
assert!(
WorkspacePathError::EffectiveRootOutsideAllowed {
roots_display: "/tmp".into(),
}
.is_policy_denied()
);
}
#[test]
fn is_policy_denied_false_for_input_shape_errors() {
assert!(!WorkspacePathError::EmptyPath.is_policy_denied());
assert!(!WorkspacePathError::WorkspaceSetPathEmpty.is_policy_denied());
assert!(!WorkspacePathError::WebEffectiveWorkspaceUnset.is_policy_denied());
assert!(
!WorkspacePathError::WebRelSubpathTooLong {
max: WEB_WORKSPACE_REL_SUBPATH_MAX_BYTES,
}
.is_policy_denied()
);
}
}
/// 路径策略与属性测试（与 golden 互补：随机探索分量边界）。
#[cfg(test)]
mod proptest_workspace_paths {
use super::*;
use proptest::prelude::*;
use tempfile::tempdir;
/// 生成形如 `/seg/seg/...` 的绝对路径（不含 `..`，便于构造子孙关系）。
fn arb_unix_abs_components(max_depth: usize) -> impl Strategy<Value = PathBuf> {
prop::collection::vec("[a-z0-9]{1,10}", 1..=max_depth).prop_map(|segments| {
let mut p = PathBuf::from("/");
for s in segments {
p.push(s.as_str());
}
p
})
}
#[test]
fn proptest_ensure_within_descendants_and_rejects_sibling_branch() {
let mut runner = proptest::test_runner::TestRunner::new(ProptestConfig::with_cases(128));
let strategy = (arb_unix_abs_components(4), 1usize..5usize, 0usize..6usize).prop_flat_map(
|(base, root_depth, extra)| {
let mut root = base.clone();
for i in 0..root_depth {
root.push(format!("r{i}"));
}
let mut candidate = root.clone();
for i in 0..extra {
candidate.push(format!("d{i}"));
}
// 与 root 同祖先 base 的另一分支：不得被 `starts_with(root)` 误判为根内。
let sibling_path = {
let mut s = base;
s.push("side_branch_not_under_root");
s
};
Just((root, candidate, sibling_path))
},
);
runner
.run(&strategy, |(root, candidate, sibling_path)| {
prop_assert!(ensure_canonical_within_root(&candidate, &root).is_ok());
prop_assert!(ensure_canonical_within_root(&sibling_path, &root).is_err());
Ok(())
})
.unwrap();
}
#[test]
fn proptest_absolutize_relative_stays_under_temp_root() {
let mut runner = proptest::test_runner::TestRunner::new(ProptestConfig::with_cases(64));
let dir = tempdir().expect("tempdir");
let root = dir.path().canonicalize().expect("canonical temp root");
let strategy = prop::collection::vec("[a-z0-9]{1,12}", 1..10);
runner
.run(&strategy, |segments| {
let mut rel = String::new();
for (i, s) in segments.iter().enumerate() {
if i > 0 {
rel.push('/');
}
rel.push_str(s);
let nested = root.join(&rel);
std::fs::create_dir_all(&nested).expect("mkdir");
}
let got = absolutize_relative_under_root(&root, &rel).expect("absolutize");
prop_assert!(got.starts_with(&root), "got={got:?} root={root:?}");
Ok(())
})
.unwrap();
}
#[test]
fn proptest_allowed_roots_matches_tempdir() {
let mut runner = proptest::test_runner::TestRunner::new(ProptestConfig::with_cases(32));
let strategy = prop::collection::vec("[a-z0-9]{1,8}", 1..6);
runner
.run(&strategy, |subdirs| {
let dir = tempdir().expect("tempdir");
let canon_root = dir.path().canonicalize().expect("canon");
let roots = vec![canon_root.clone()];
let mut target = canon_root.clone();
for s in &subdirs {
target.push(s);
}
std::fs::create_dir_all(&target).expect("mkdir");
let canon_target = target.canonicalize().expect("canon target");
prop_assert!(is_within_allowed_roots(&canon_target, &roots));
Ok(())
})
.unwrap();
}
#[test]
fn proptest_absolutize_rejects_parent_escape() {
let dir = tempdir().expect("tempdir");
let root = dir.path().canonicalize().expect("canon");
let res = absolutize_relative_under_root(&root, "../../../etc/passwd");
assert!(res.is_err());
assert_eq!(res.expect_err("escape").kind(), "outside_workspace_root");
}
}