feat(files-mcp): scaffold package + path_guard sandbox
Adds the files_mcp/ sibling package and core/path_guard.py which resolves arbitrary input paths and rejects any that escape the FILES_MCP_ROOT sandbox (handles .., ~, symlinks, NUL, empty). Adds tests/unit/test_path_guard.py covering accept, escape modes, symlink escape, and _init_root failure paths. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,29 @@
|
||||
# coding=utf-8
|
||||
"""
|
||||
@Time :2026/6/30
|
||||
@Author :tao.chen
|
||||
"""
|
||||
from pathlib import Path
|
||||
|
||||
from files_mcp.core import path_guard
|
||||
|
||||
# Fail fast at process start if FILES_MCP_ROOT is unset / bad. During tests
|
||||
# the root is injected via set_root_for_testing(), so tolerate the env var
|
||||
# being unset here rather than crashing at import/collection time.
|
||||
_ROOT: Path | None
|
||||
try:
|
||||
_ROOT = path_guard._init_root()
|
||||
except RuntimeError:
|
||||
_ROOT = None
|
||||
|
||||
# Re-export the FastAPI app once the sandbox root has been initialized.
|
||||
# server.py is created in a later task; importing it lazily keeps this
|
||||
# package importable (and unit-testable) before that task lands.
|
||||
try: # pragma: no cover - exercised once server.py exists
|
||||
from files_mcp.server import app # noqa: E402
|
||||
except ImportError:
|
||||
app = None # type: ignore[assignment]
|
||||
|
||||
ROOT: Path | None = _ROOT
|
||||
|
||||
__all__ = ["app", "ROOT"]
|
||||
@@ -0,0 +1,5 @@
|
||||
# coding=utf-8
|
||||
"""
|
||||
@Time :2026/6/30
|
||||
@Author :tao.chen
|
||||
"""
|
||||
@@ -0,0 +1,83 @@
|
||||
# coding=utf-8
|
||||
"""
|
||||
@Time :2026/6/30
|
||||
@Author :tao.chen
|
||||
|
||||
Sandbox: every user-supplied path is resolve()d and checked against a
|
||||
single frozen root directory. resolve() follows symlinks and normalizes
|
||||
.., ~, and duplicate slashes, so a single relative_to() check is enough.
|
||||
|
||||
Relative inputs are anchored at the root: a bare "inside.txt" means
|
||||
"<root>/inside.txt", not "<cwd>/inside.txt". Absolute paths and paths
|
||||
that already contain a separator (e.g. "~/x") are left intact so ~ and
|
||||
leading-slash inputs behave as expected.
|
||||
"""
|
||||
import os
|
||||
from pathlib import Path
|
||||
|
||||
from common.logging import logger
|
||||
|
||||
_ROOT: Path | None = None
|
||||
|
||||
|
||||
def _init_root() -> Path:
|
||||
"""Resolve and freeze the sandbox root from FILES_MCP_ROOT.
|
||||
|
||||
Called once at package import time. Raises RuntimeError if the env
|
||||
var is unset, points at a missing path, or points at a non-directory.
|
||||
Idempotent: subsequent calls return the cached root.
|
||||
"""
|
||||
global _ROOT
|
||||
if _ROOT is not None:
|
||||
return _ROOT
|
||||
raw = os.environ.get("FILES_MCP_ROOT")
|
||||
if not raw:
|
||||
raise RuntimeError(
|
||||
"FILES_MCP_ROOT env var is required (e.g. export "
|
||||
"FILES_MCP_ROOT=/srv/files-mcp-sandbox)"
|
||||
)
|
||||
p = Path(raw).expanduser().resolve()
|
||||
if not p.is_dir():
|
||||
raise RuntimeError(
|
||||
f"FILES_MCP_ROOT={raw!r} resolves to {p} which is not an existing directory"
|
||||
)
|
||||
_ROOT = p
|
||||
logger.info(f"Files MCP sandbox root: {_ROOT}")
|
||||
return _ROOT
|
||||
|
||||
|
||||
def resolve_and_check(path: str) -> Path:
|
||||
"""Resolve an arbitrary input path and verify it lies inside the sandbox.
|
||||
|
||||
Raises ValueError if the path is empty, contains NUL, or escapes the
|
||||
root after symlink/.. resolution.
|
||||
"""
|
||||
if _ROOT is None:
|
||||
_init_root()
|
||||
assert _ROOT is not None
|
||||
if not path:
|
||||
raise ValueError("path is empty")
|
||||
if "\x00" in path:
|
||||
raise ValueError("path contains NUL byte")
|
||||
# Anchor bare relative names at the sandbox root so "inside.txt" means
|
||||
# "<root>/inside.txt", not "<cwd>/inside.txt". Keep user home ("~...") and
|
||||
# absolute paths untouched so they still resolve / reject as expected.
|
||||
if not os.path.isabs(path) and not path.startswith("~"):
|
||||
candidate = _ROOT / path
|
||||
else:
|
||||
candidate = Path(path)
|
||||
p = candidate.expanduser().resolve()
|
||||
try:
|
||||
p.relative_to(_ROOT)
|
||||
except ValueError:
|
||||
raise ValueError(
|
||||
f"path escapes sandbox root {_ROOT}: got {p}"
|
||||
) from None
|
||||
return p
|
||||
|
||||
|
||||
def set_root_for_testing(p: Path) -> None:
|
||||
"""Test-only escape hatch. Never call from production code."""
|
||||
global _ROOT
|
||||
_ROOT = p.resolve()
|
||||
os.environ["FILES_MCP_ROOT"] = str(_ROOT)
|
||||
@@ -0,0 +1,84 @@
|
||||
# coding=utf-8
|
||||
"""
|
||||
@Time :2026/6/30
|
||||
@Author :tao.chen
|
||||
"""
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from files_mcp.core import path_guard
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def sandbox_root(tmp_path, monkeypatch):
|
||||
"""Fresh tmp_path-based sandbox root for each test.
|
||||
|
||||
Pre-populates a regular file, a nested file inside a subdir, and a
|
||||
symlink that points outside the sandbox (for the symlink-escape test).
|
||||
"""
|
||||
monkeypatch.setenv("FILES_MCP_ROOT", str(tmp_path))
|
||||
path_guard.set_root_for_testing(tmp_path.resolve())
|
||||
(tmp_path / "inside.txt").write_text("hi", encoding="utf-8")
|
||||
(tmp_path / "subdir").mkdir()
|
||||
(tmp_path / "subdir" / "nested.txt").write_text("nested", encoding="utf-8")
|
||||
(tmp_path / "escape_link").symlink_to("/etc/passwd")
|
||||
return tmp_path.resolve()
|
||||
|
||||
|
||||
class TestResolveAndCheck:
|
||||
def test_accepts_inside_path(self, sandbox_root):
|
||||
p = path_guard.resolve_and_check("inside.txt")
|
||||
assert p == sandbox_root / "inside.txt"
|
||||
|
||||
def test_accepts_nested_path(self, sandbox_root):
|
||||
p = path_guard.resolve_and_check("subdir/nested.txt")
|
||||
assert p == sandbox_root / "subdir" / "nested.txt"
|
||||
|
||||
def test_accepts_dot_segments(self, sandbox_root):
|
||||
p = path_guard.resolve_and_check("subdir/../inside.txt")
|
||||
assert p == sandbox_root / "inside.txt"
|
||||
|
||||
def test_rejects_double_dot_escape(self, sandbox_root):
|
||||
with pytest.raises(ValueError, match="escapes sandbox"):
|
||||
path_guard.resolve_and_check("../etc/passwd")
|
||||
|
||||
def test_rejects_absolute_outside_path(self, sandbox_root):
|
||||
with pytest.raises(ValueError, match="escapes sandbox"):
|
||||
path_guard.resolve_and_check("/etc/passwd")
|
||||
|
||||
def test_rejects_mixed_escape(self, sandbox_root):
|
||||
with pytest.raises(ValueError, match="escapes sandbox"):
|
||||
path_guard.resolve_and_check("subdir/../../etc/passwd")
|
||||
|
||||
def test_rejects_symlink_escape(self, sandbox_root):
|
||||
# escape_link -> /etc/passwd; resolve() follows the link first.
|
||||
with pytest.raises(ValueError, match="escapes sandbox"):
|
||||
path_guard.resolve_and_check("escape_link")
|
||||
|
||||
def test_rejects_empty(self, sandbox_root):
|
||||
with pytest.raises(ValueError, match="path is empty"):
|
||||
path_guard.resolve_and_check("")
|
||||
|
||||
def test_rejects_nul_byte(self, sandbox_root):
|
||||
with pytest.raises(ValueError, match="NUL byte"):
|
||||
path_guard.resolve_and_check("foo\x00bar")
|
||||
|
||||
|
||||
class TestInitRoot:
|
||||
def test_init_root_raises_if_unset(self, monkeypatch):
|
||||
monkeypatch.delenv("FILES_MCP_ROOT", raising=False)
|
||||
path_guard._ROOT = None
|
||||
with pytest.raises(RuntimeError, match="FILES_MCP_ROOT"):
|
||||
path_guard._init_root()
|
||||
|
||||
def test_init_root_raises_if_not_directory(self, tmp_path, monkeypatch):
|
||||
monkeypatch.setenv("FILES_MCP_ROOT", str(tmp_path / "missing"))
|
||||
path_guard._ROOT = None
|
||||
with pytest.raises(RuntimeError, match="not an existing directory"):
|
||||
path_guard._init_root()
|
||||
|
||||
def test_init_root_is_idempotent(self, sandbox_root):
|
||||
first = path_guard._ROOT
|
||||
path_guard._init_root()
|
||||
assert path_guard._ROOT == first
|
||||
Reference in New Issue
Block a user