More TOCTOU, update to tests for jinjaturtle

enroll/fsutil.py

@@ -7,7 +7,13 @@ import stat
 from typing import Tuple
 
 
-def open_no_follow_path(path: str, *, write: bool = False, mode: int = 0o600) -> int:
+def open_no_follow_path(
+    path: str,
+    *,
+    write: bool = False,
+    mode: int = 0o600,
+    directory: bool = False,
+) -> int:
     """Open ``path`` without following a symlink in *any* path component.
 
     ``O_NOFOLLOW`` only protects the final component of a path. A regular
@@ -42,10 +48,21 @@ def open_no_follow_path(path: str, *, write: bool = False, mode: int = 0o600) ->
     o_directory = getattr(os, "O_DIRECTORY", 0)
     o_path = getattr(os, "O_PATH", 0)
 
+    if write and directory:
+        raise ValueError("directory=True cannot be combined with write=True")
+
     if write:
         final_flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL | cloexec | nofollow
+    elif directory and o_path:
+        # O_PATH|O_NOFOLLOW opens a final symlink as the symlink object itself
+        # on Linux, allowing inspect_dir_no_follow() to reject it via fstat().
+        # O_RDONLY|O_DIRECTORY|O_NOFOLLOW can still follow a symlink-to-dir on
+        # some kernels/filesystems.
+        final_flags = o_path | cloexec | nofollow
     else:
         final_flags = os.O_RDONLY | cloexec | nofollow
+        if directory:
+            final_flags |= o_directory
 
     supports_openat = bool(
         o_directory and nofollow and os.open in getattr(os, "supports_dir_fd", set())
@@ -126,6 +143,40 @@ def open_no_follow_path(path: str, *, write: bool = False, mode: int = 0o600) ->
         os.close(dir_fd)
 
 
+def inspect_dir_no_follow(path: str) -> os.stat_result:
+    """Return fstat() metadata for a directory opened without following symlinks.
+
+    Directory metadata capture must have the same TOCTOU properties as file
+    capture: inspect the exact object reached through a no-follow descriptor,
+    and reject symlink components anywhere in the path.  Path-based
+    ``os.stat()`` / ``os.path.isdir()`` checks can be swapped between check and
+    use when an include root is attacker-writable; this helper keeps the check
+    bound to the opened descriptor.
+    """
+
+    fd = open_no_follow_path(path, directory=True)
+    try:
+        st = os.fstat(fd)
+        if stat.S_ISLNK(st.st_mode):
+            raise OSError(errno.ELOOP, "symlinked directory path", path)
+        if not stat.S_ISDIR(st.st_mode):
+            raise OSError(errno.ENOTDIR, "not a directory", path)
+        return st
+    finally:
+        os.close(fd)
+
+
+def stat_dir_triplet(path: str) -> Tuple[str, str, str]:
+    """Return (owner, group, mode) for a safely-opened directory path.
+
+    Unlike :func:`stat_triplet`, this refuses final symlinks and symlinked
+    parent components, and derives metadata from the directory descriptor that
+    passed those checks.
+    """
+
+    return stat_triplet_from_stat(inspect_dir_no_follow(path))
+
+
 def path_has_symlink_component(path: str) -> bool:
     """Return True if any existing component of *path* is a symlink.
 

enroll/harvest.py

@@ -12,7 +12,7 @@ from typing import Any, Dict, List, Optional, Set, Tuple
 
 from . import accounts as _accounts
 from . import systemd as _systemd
-from .fsutil import stat_triplet
+from .fsutil import stat_dir_triplet
 from .platform import detect_platform, get_backend
 from .ignore import IgnorePolicy
 from .harvest_safety import ensure_private_empty_dir, prepare_new_private_dir
@@ -118,7 +118,7 @@ def _merge_parent_dirs(
             continue
 
         try:
-            owner, group, mode = stat_triplet(dpath)
+            owner, group, mode = stat_dir_triplet(dpath)
         except OSError:
             continue
 

enroll/harvest_collectors/paths.py

@@ -254,7 +254,7 @@ class ExtraPathsCollector(HarvestCollector):
                         deny = None
                 if not deny:
                     try:
-                        owner, group, mode = h.stat_triplet(dirpath)
+                        owner, group, mode = h.stat_dir_triplet(dirpath)
                         self.managed_dirs.append(
                             ManagedDir(
                                 path=dirpath,

enroll/ignore.py

@@ -1,14 +1,14 @@
 from __future__ import annotations
 
-import fnmatch
 import errno
+import fnmatch
 import os
 import re
 import stat
 from dataclasses import dataclass
 from typing import Optional
 
-from .fsutil import open_no_follow_path
+from .fsutil import inspect_dir_no_follow, open_no_follow_path
 
 
 DEFAULT_DENY_GLOBS = [
@@ -407,16 +407,14 @@ class IgnorePolicy:
                     return "denied_path"
 
         try:
-            os.stat(path, follow_symlinks=True)
+            inspect_dir_no_follow(path)
-        except OSError:
+        except OSError as exc:
+            if exc.errno == errno.ELOOP:
+                return "symlink"
+            if exc.errno == errno.ENOTDIR:
+                return "not_directory"
             return "unreadable"
 
-        if os.path.islink(path):
-            return "symlink"
-
-        if not os.path.isdir(path):
-            return "not_directory"
-
         return None
 
     def deny_reason_link(self, path: str) -> Optional[str]:

tests.sh

@@ -228,6 +228,7 @@ ensure_jinjaturtle() {
 
   # Clone git repo
   run git clone https://git.mig5.net/mig5/jinjaturtle /tmp/jinjaturtle
+  cd /tmp/jinjaturtle && run poetry install --with dev
   cd /tmp/jinjaturtle && run poetry build
   cd /tmp/jinjaturtle && run poetry run pyproject-appimage --output /usr/bin/jinjaturtle
 }
@@ -319,12 +320,6 @@ run_ansible_jinjaturtle_variant() {
 }
 
 run_jinjaturtle_manifest_tests() {
-  if is_rpm_family ; then
-    section "JinjaTurtle integration matrix"
-    printf 'Skipping JinjaTurtle package integration on RPM-family CI;\n'
-    return
-  fi
-
   ensure_jinjaturtle
   require_cmd jinjaturtle "Install JinjaTurtle before running the JinjaTurtle integration matrix."
 

tests/test_fsutil.py

@@ -4,7 +4,7 @@ import hashlib
 import os
 from pathlib import Path
 
-from enroll.fsutil import file_md5, stat_triplet
+from enroll.fsutil import file_md5, stat_dir_triplet, stat_triplet
 
 
 def test_file_md5_matches_hashlib(tmp_path: Path):
@@ -74,3 +74,46 @@ def test_open_no_follow_path_refuses_symlinked_leaf(tmp_path: Path):
         raise AssertionError("expected OSError for symlinked leaf")
     except OSError as e:
         assert e.errno == errno.ELOOP
+
+
+def test_stat_dir_triplet_reports_directory_mode_without_following(tmp_path: Path):
+    d = tmp_path / "dir"
+    d.mkdir()
+    os.chmod(d, 0o750)
+
+    owner, group, mode = stat_dir_triplet(str(d))
+    assert mode == "0750"
+    assert owner
+    assert group
+
+
+def test_stat_dir_triplet_refuses_symlinked_parent(tmp_path: Path):
+    import errno
+
+    real = tmp_path / "real"
+    real.mkdir()
+    child = real / "child"
+    child.mkdir()
+    link = tmp_path / "link"
+    link.symlink_to(real, target_is_directory=True)
+
+    try:
+        stat_dir_triplet(str(link / "child"))
+        raise AssertionError("expected OSError for symlinked parent")
+    except OSError as e:
+        assert e.errno == errno.ELOOP
+
+
+def test_stat_dir_triplet_refuses_final_symlink(tmp_path: Path):
+    import errno
+
+    real = tmp_path / "real"
+    real.mkdir()
+    link = tmp_path / "link"
+    link.symlink_to(real, target_is_directory=True)
+
+    try:
+        stat_dir_triplet(str(link))
+        raise AssertionError("expected OSError for symlinked leaf")
+    except OSError as e:
+        assert e.errno == errno.ELOOP

tests/test_harvest.py

@@ -256,6 +256,7 @@ def test_harvest_dedup_manual_packages_and_builds_etc_custom(
         return ("root", "root", "0644")
 
     monkeypatch.setattr(harvest, "stat_triplet", fake_stat_triplet)
+    monkeypatch.setattr(harvest, "stat_dir_triplet", fake_stat_triplet)
     monkeypatch.setattr(capture, "stat_triplet", fake_stat_triplet)
 
     # Avoid needing source files on disk by implementing our own bundle copier
@@ -400,6 +401,7 @@ def test_shared_cron_snippet_prefers_matching_role_over_lexicographic(
     monkeypatch.setattr(harvest, "get_backend", lambda info=None: backend)
 
     monkeypatch.setattr(harvest, "stat_triplet", lambda p: ("root", "root", "0644"))
+    monkeypatch.setattr(harvest, "stat_dir_triplet", lambda p: ("root", "root", "0755"))
     monkeypatch.setattr(capture, "stat_triplet", lambda p: ("root", "root", "0644"))
     monkeypatch.setattr(harvest, "collect_non_system_users", lambda: [])
 

tests/test_harvest_collectors.py

@@ -285,7 +285,7 @@ def test_extra_paths_collector_records_dirs_files_notes_and_excludes(
         )
         return True
 
-    monkeypatch.setattr(paths.h, "stat_triplet", fake_stat_triplet)
+    monkeypatch.setattr(paths.h, "stat_dir_triplet", fake_stat_triplet)
     monkeypatch.setattr(paths, "capture_file", lambda *a, **kw: fake_capture_file(**kw))
 
     ctx = _context(
@@ -321,7 +321,7 @@ def test_extra_paths_collector_skips_already_captured_files(monkeypatch, tmp_pat
     file_path.write_text("ok", encoding="utf-8")
     calls: list[str] = []
 
-    monkeypatch.setattr(paths.h, "stat_triplet", lambda p: ("root", "root", "0755"))
+    monkeypatch.setattr(paths.h, "stat_dir_triplet", lambda p: ("root", "root", "0755"))
     monkeypatch.setattr(
         paths, "capture_file", lambda *a, **kw: calls.append(kw["abs_path"]) or True
     )

tests/test_ignore_dir.py

@@ -39,8 +39,12 @@ def test_deny_reason_dir_behaviour(tmp_path: Path):
     link = tmp_path / "link"
     link.symlink_to(d)
 
+    parent_link = tmp_path / "parent_link"
+    parent_link.symlink_to(tmp_path, target_is_directory=True)
+
     assert pol.deny_reason_dir(str(d)) is None
     assert pol.deny_reason_dir(str(link)) == "symlink"
+    assert pol.deny_reason_dir(str(parent_link / "dir")) == "symlink"
     assert pol.deny_reason_dir(str(f)) == "not_directory"
 
     # Denied by glob.