diff --git a/wizard-work/beam/pathguard.py b/wizard-work/beam/pathguard.py new file mode 100644 index 0000000..7b27ede --- /dev/null +++ b/wizard-work/beam/pathguard.py @@ -0,0 +1,83 @@ +"""Safe path join: prevent traversal outside a base directory. + +safe_join(base, *parts) returns an absolute path that is guaranteed +to live inside base (resolved). Raises PathTraversalError on any attempt +to escape (absolute parts, '..', symlinks that resolve outside, etc.). +""" +from __future__ import annotations + +import os +from pathlib import Path, PurePath + + +class PathTraversalError(ValueError): + """Raised when a joined path escapes the base directory.""" + + +def _is_under(child: Path, parent: Path) -> bool: + try: + child.relative_to(parent) + return True + except ValueError: + return False + + +def safe_join(base, *parts) -> str: + """Join *parts onto base, rejecting anything that escapes base. + + - base must be an existing directory (or creatable? we require existing). + - Each part is treated as a relative path; absolute parts and '..' + segments that escape are rejected. + - Symlinks are resolved; if a symlink points outside base, rejected. + - Empty parts are ignored. + - Returns a normalized absolute string path. + """ + base_path = Path(base).resolve(strict=True) + if not base_path.is_dir(): + raise PathTraversalError(f"base is not a directory: {base}") + + # Reject absolute parts up front (they would replace base semantics). + for p in parts: + if p is None: + continue + pp = PurePath(p) + if pp.is_absolute(): + raise PathTraversalError(f"absolute part not allowed: {p!r}") + # Also reject drive/root markers on any platform. + if pp.drive or pp.root: + raise PathTraversalError(f"rooted part not allowed: {p!r}") + + # Build candidate without resolving first, to detect traversal segments. + candidate = base_path + for p in parts: + if p is None or str(p) == "": + continue + # Normalize separators; reject NUL. + s = str(p) + if "\x00" in s: + raise PathTraversalError("NUL byte in part") + candidate = candidate.joinpath(s) + + # Resolve; if a component doesn't exist, resolve still normalizes lexicaly + # for the non-existent tail, but existing symlinks are resolved. + try: + resolved = candidate.resolve(strict=False) + except (OSError, RuntimeError) as e: + raise PathTraversalError(f"cannot resolve path: {e}") from e + + if not _is_under(resolved, base_path): + raise PathTraversalError(f"path escapes base: {resolved}") + + # Extra check: ensure no intermediate '..' escaped even if final lands inside + # (e.g. symlink tricks). Walk lexical parts. + lexical = base_path + for p in parts: + if p is None or str(p) == "": + continue + lexical = lexical.joinpath(str(p)) + # lexical may contain '..'; compute normalized + norm = os.path.normpath(str(lexical)) + if not _is_under(Path(norm), base_path): + raise PathTraversalError(f"lexical path escapes base: {norm}") + + return str(resolved) \ No newline at end of file diff --git a/wizard-work/beam/test_pathguard.py b/wizard-work/beam/test_pathguard.py new file mode 100644 index 0000000..126b04e --- /dev/null +++ b/wizard-work/beam/test_pathguard.py @@ -0,0 +1,66 @@ +import os +import tempfile +import pytest +from pathguard import safe_join, PathTraversalError + + +def test_basic_join(): + with tempfile.TemporaryDirectory() as d: + out = safe_join(d, "a", "b.txt") + assert out == os.path.join(os.path.realpath(d), "a", "b.txt") + + +def test_dotdot_escape_rejected(): + with tempfile.TemporaryDirectory() as d: + with pytest.raises(PathTraversalError): + safe_join(d, "..", "etc", "passwd") + + +def test_absolute_part_rejected(): + with tempfile.TemporaryDirectory() as d: + with pytest.raises(PathTraversalError): + safe_join(d, "/etc/passwd") + + +def test_symlink_escape_rejected(): + with tempfile.TemporaryDirectory() as d: + base = os.path.join(d, "base") + os.makedirs(base) + target = os.path.join(d, "secret.txt") + with open(target, "w") as f: + f.write("secret") + link = os.path.join(base, "link") + os.symlink(target, link) + # reading through link to outside must be rejected when resolved? + # safe_join resolves; link itself is inside base by name but points outside. + # Our function resolves symlinks and checks containment -> should reject. + with pytest.raises(PathTraversalError): + safe_join(base, "link") + + +def test_valid_nested(): + with tempfile.TemporaryDirectory() as d: + sub = os.path.join(d, "sub") + os.makedirs(sub) + out = safe_join(d, "sub", "file.txt") + assert out.startswith(os.path.realpath(d) + os.sep) + + +def test_empty_parts_ignored(): + with tempfile.TemporaryDirectory() as d: + out = safe_join(d, "", None, "x") + assert out.endswith("x") + + +def test_nul_rejected(): + with tempfile.TemporaryDirectory() as d: + with pytest.raises(PathTraversalError): + safe_join(d, "a\x00b") + + +def test_base_not_dir(): + with tempfile.TemporaryDirectory() as d: + f = os.path.join(d, "f") + open(f, "w").close() + with pytest.raises(PathTraversalError): + safe_join(f, "x")