Fix tests

src/jinjaturtle/output_safety.py

@@ -2,11 +2,12 @@ from __future__ import annotations
 
 """Safer file-output helpers for the JinjaTurtle CLI.
 
-The CLI is often used by administrators.  A plain Path.write_text() follows a
+The CLI is often used by administrators. A plain Path.write_text() follows a
 final-path symlink and can therefore be dangerous when a root-run invocation
-writes into an attacker-writable tree.  These helpers validate path components,
+writes into an attacker-writable tree. These helpers validate path components,
-write through a private temporary file in the target directory, and replace the
+refuse symlink parents, reject root-run output through untrusted parent
-final path atomically.  Existing final-path symlinks are refused rather than
+components, write through a temporary file in the target directory, and replace
+the final path atomically. Existing final-path symlinks are refused rather than
 followed.
 """
 
@@ -20,8 +21,111 @@ class OutputPathError(OSError):
     """Raised when a requested output path is unsafe."""
 
 
+# Keep a reference to the real euid getter. Tests can monkeypatch
+# _effective_uid directly without changing process-wide os.geteuid behaviour.
+_OS_GETEUID = getattr(os, "geteuid", None)
+
+
+def _effective_uid() -> int | None:
+    if _OS_GETEUID is None:
+        return None
+    try:
+        return int(_OS_GETEUID())
+    except OSError:
+        return None
+
+
 def _absolute(path: Path) -> Path:
-    return path if path.is_absolute() else Path.cwd() / path
+    expanded = path.expanduser()
+    return expanded if expanded.is_absolute() else Path.cwd() / expanded
+
+
+def _chmod_private(path: Path) -> None:
+    try:
+        os.chmod(path, 0o700)
+    except OSError:
+        # Best-effort; mkdir(mode=0o700) is already used for normal filesystems.
+        pass
+
+
+def _assert_trusted_root_parent(path: Path, st: os.stat_result) -> None:
+    """Reject root-run output through attacker-controlled parent directories.
+
+    A root-run JinjaTurtle process may write files that an administrator later
+    applies as configuration-management input. Parent directories controlled by
+    an unprivileged user are therefore not acceptable output anchors. Root-owned
+    sticky shared directories such as /tmp are allowed as a boundary, but any
+    existing child below them must be root-owned and not writable by group/other.
+    """
+
+    if _effective_uid() != 0:
+        return
+    if not stat.S_ISDIR(st.st_mode):
+        raise OutputPathError(f"output parent is not a directory: {path}")
+    if st.st_uid != 0:
+        raise OutputPathError(
+            f"output parent is not owned by root; refusing root-run output: {path}"
+        )
+    writable_by_group_or_other = st.st_mode & (stat.S_IWGRP | stat.S_IWOTH)
+    sticky = st.st_mode & stat.S_ISVTX
+    if writable_by_group_or_other and not sticky:
+        raise OutputPathError(
+            "output parent is writable by group/other; "
+            f"refusing root-run output: {path}"
+        )
+
+
+def _assert_existing_directory_component(path: Path) -> None:
+    try:
+        st = path.lstat()
+    except OSError as exc:
+        raise OutputPathError(f"unable to inspect output parent: {path}") from exc
+    if stat.S_ISLNK(st.st_mode):
+        raise OutputPathError(f"refusing to use symlink parent: {path}")
+    if not stat.S_ISDIR(st.st_mode):
+        raise OutputPathError(f"output parent is not a directory: {path}")
+    _assert_trusted_root_parent(path, st)
+
+
+def _mkdir_safe_dir_tree(path: Path) -> Path:
+    """Create/validate a directory tree one component at a time.
+
+    pathlib.mkdir(parents=True) can traverse a symlink inserted after a parent
+    pre-check. Walking one component at a time keeps every existing component
+    checked before it is used, and newly-created components are immediately
+    re-inspected.
+    """
+
+    out = _absolute(path)
+    parts = out.parts
+    if not parts:
+        return out
+
+    if out.is_absolute():
+        cur = Path(parts[0])
+        rest = parts[1:]
+        _assert_existing_directory_component(cur)
+    else:
+        # _absolute() currently always returns an absolute path, but keep this
+        # branch for clarity if that helper is ever relaxed.
+        cur = Path.cwd()
+        rest = parts
+        _assert_existing_directory_component(cur)
+
+    for part in rest:
+        cur = cur / part
+        if os.path.lexists(cur):
+            _assert_existing_directory_component(cur)
+            continue
+        try:
+            os.mkdir(cur, 0o700)
+        except FileExistsError:
+            _assert_existing_directory_component(cur)
+            continue
+        _chmod_private(cur)
+        _assert_existing_directory_component(cur)
+
+    return out
 
 
 def _check_existing_output_file(path: Path) -> None:
@@ -36,65 +140,36 @@ def _check_existing_output_file(path: Path) -> None:
 
 
 def _check_parent_components(parent: Path) -> None:
-    """Require every existing parent component to be a real directory."""
+    """Require every existing parent component to be a trusted real directory."""
 
-    parent = _absolute(parent)
+    _mkdir_safe_dir_tree(parent)
-    parts = parent.parts
-    if not parts:
-        return
-
-    cur = Path(parts[0])
-    for part in parts[1:]:
-        cur = cur / part
-        try:
-            st = cur.lstat()
-        except FileNotFoundError as exc:
-            raise OutputPathError(f"output parent does not exist: {cur}") from exc
-        if stat.S_ISLNK(st.st_mode):
-            raise OutputPathError(f"refusing to use symlink parent: {cur}")
-        if not stat.S_ISDIR(st.st_mode):
-            raise OutputPathError(f"output parent is not a directory: {cur}")
 
 
 def ensure_safe_directory(path: Path) -> None:
-    """Create or validate a directory tree without accepting symlinks."""
+    """Create or validate a directory tree without accepting unsafe parents."""
 
-    path = _absolute(path)
+    _mkdir_safe_dir_tree(path)
-    parts = path.parts
-    if not parts:
-        return
-
-    cur = Path(parts[0])
-    for part in parts[1:]:
-        cur = cur / part
-        try:
-            st = cur.lstat()
-        except FileNotFoundError:
-            cur.mkdir(mode=0o700)
-            st = cur.lstat()
-        if stat.S_ISLNK(st.st_mode):
-            raise OutputPathError(f"refusing to use symlink directory: {cur}")
-        if not stat.S_ISDIR(st.st_mode):
-            raise OutputPathError(f"output path is not a directory: {cur}")
 
 
 def write_text_safely(path: Path, text: str, *, encoding: str = "utf-8") -> None:
     """Write text without following a final-path symlink.
 
-    The target's parent must already exist and every parent component must be a
+    The target's parent is created/validated component by component. Existing
-    real directory.  The write is completed with os.replace(), which atomically
+    parent symlinks are refused for every user. When running as root, existing
-    swaps the final directory entry and does not dereference a final symlink.
+    parent components must also be root-owned and not writable by group/other,
+    except for sticky shared boundaries such as /tmp. Existing final-path
+    symlinks are refused rather than followed.
     """
 
     path = _absolute(path)
-    _check_parent_components(path.parent)
+    parent = _mkdir_safe_dir_tree(path.parent)
     _check_existing_output_file(path)
 
     fd = -1
     tmp_name: str | None = None
     try:
         fd, tmp_name = tempfile.mkstemp(
-            prefix=f".{path.name}.", suffix=".tmp", dir=str(path.parent), text=True
+            prefix=f".{path.name}.", suffix=".tmp", dir=str(parent), text=True
         )
         with os.fdopen(fd, "w", encoding=encoding) as f:
             fd = -1
@@ -102,6 +177,7 @@ def write_text_safely(path: Path, text: str, *, encoding: str = "utf-8") -> None
             f.flush()
             os.fsync(f.fileno())
         os.chmod(tmp_name, 0o600)
+        _check_parent_components(path.parent)
         _check_existing_output_file(path)
         os.replace(tmp_name, path)
         tmp_name = None

tests/test_security_hardening.py

@@ -118,3 +118,19 @@ def test_cli_reports_unsafe_output_path_without_overwriting_symlink(tmp_path: Pa
 
     assert exit_code == 2
     assert target.read_text(encoding="utf-8") == "keep\n"
+
+
+def test_cli_refuses_root_output_through_group_writable_parent(
+    tmp_path: Path, monkeypatch
+):
+    from jinjaturtle import output_safety
+
+    unsafe_dir = tmp_path / "unsafe"
+    unsafe_dir.mkdir()
+    unsafe_dir.chmod(0o777)
+    monkeypatch.setattr(output_safety, "_effective_uid", lambda: 0)
+
+    with pytest.raises(OutputPathError):
+        write_text_safely(unsafe_dir / "out.yml", "data\n")
+
+    assert not (unsafe_dir / "out.yml").exists()