More TOCTOU, update to tests for jinjaturtle
Some checks failed
CI / test (push) Successful in 48s
CI / test (almalinux, docker.io/library/almalinux:9, python3.11) (push) Failing after 2m32s
CI / test (debian, docker.io/library/debian:13, python3) (push) Failing after 3m0s
Lint / test (push) Successful in 44s

This commit is contained in:
Miguel Jacq 2026-06-29 14:30:07 +10:00
parent 56ae883948
commit 3c1e08bdde
Signed by: mig5
GPG key ID: 03906B4110AAD3B8
9 changed files with 116 additions and 17 deletions

View file

@ -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.

View file

@ -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

View file

@ -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,

View file

@ -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,15 +407,13 @@ class IgnorePolicy:
return "denied_path"
try:
os.stat(path, follow_symlinks=True)
except OSError:
return "unreadable"
if os.path.islink(path):
inspect_dir_no_follow(path)
except OSError as exc:
if exc.errno == errno.ELOOP:
return "symlink"
if not os.path.isdir(path):
if exc.errno == errno.ENOTDIR:
return "not_directory"
return "unreadable"
return None

View file

@ -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
}

View file

@ -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

View file

@ -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: [])

View file

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

View file

@ -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.