beam: add pathguard safe-join module with tests

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)

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")