Compare commits
No commits in common. "3c1e08bddefcb9cd50699ef5275009af87e79008" and "76aafcc08dfa6f27a191859c858202c394c5714e" have entirely different histories.
3c1e08bdde
...
76aafcc08d
9 changed files with 23 additions and 116 deletions
|
|
@ -7,13 +7,7 @@ import stat
|
|||
from typing import Tuple
|
||||
|
||||
|
||||
def open_no_follow_path(
|
||||
path: str,
|
||||
*,
|
||||
write: bool = False,
|
||||
mode: int = 0o600,
|
||||
directory: bool = False,
|
||||
) -> int:
|
||||
def open_no_follow_path(path: str, *, write: bool = False, mode: int = 0o600) -> int:
|
||||
"""Open ``path`` without following a symlink in *any* path component.
|
||||
|
||||
``O_NOFOLLOW`` only protects the final component of a path. A regular
|
||||
|
|
@ -48,21 +42,10 @@ def open_no_follow_path(
|
|||
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())
|
||||
|
|
@ -143,40 +126,6 @@ def open_no_follow_path(
|
|||
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.
|
||||
|
||||
|
|
|
|||
|
|
@ -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_dir_triplet
|
||||
from .fsutil import stat_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_dir_triplet(dpath)
|
||||
owner, group, mode = stat_triplet(dpath)
|
||||
except OSError:
|
||||
continue
|
||||
|
||||
|
|
|
|||
|
|
@ -254,7 +254,7 @@ class ExtraPathsCollector(HarvestCollector):
|
|||
deny = None
|
||||
if not deny:
|
||||
try:
|
||||
owner, group, mode = h.stat_dir_triplet(dirpath)
|
||||
owner, group, mode = h.stat_triplet(dirpath)
|
||||
self.managed_dirs.append(
|
||||
ManagedDir(
|
||||
path=dirpath,
|
||||
|
|
|
|||
|
|
@ -1,14 +1,14 @@
|
|||
from __future__ import annotations
|
||||
|
||||
import errno
|
||||
import fnmatch
|
||||
import errno
|
||||
import os
|
||||
import re
|
||||
import stat
|
||||
from dataclasses import dataclass
|
||||
from typing import Optional
|
||||
|
||||
from .fsutil import inspect_dir_no_follow, open_no_follow_path
|
||||
from .fsutil import open_no_follow_path
|
||||
|
||||
|
||||
DEFAULT_DENY_GLOBS = [
|
||||
|
|
@ -407,14 +407,16 @@ class IgnorePolicy:
|
|||
return "denied_path"
|
||||
|
||||
try:
|
||||
inspect_dir_no_follow(path)
|
||||
except OSError as exc:
|
||||
if exc.errno == errno.ELOOP:
|
||||
return "symlink"
|
||||
if exc.errno == errno.ENOTDIR:
|
||||
return "not_directory"
|
||||
os.stat(path, follow_symlinks=True)
|
||||
except OSError:
|
||||
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]:
|
||||
|
|
|
|||
7
tests.sh
7
tests.sh
|
|
@ -228,7 +228,6 @@ 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
|
||||
}
|
||||
|
|
@ -320,6 +319,12 @@ 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."
|
||||
|
||||
|
|
|
|||
|
|
@ -4,7 +4,7 @@ import hashlib
|
|||
import os
|
||||
from pathlib import Path
|
||||
|
||||
from enroll.fsutil import file_md5, stat_dir_triplet, stat_triplet
|
||||
from enroll.fsutil import file_md5, stat_triplet
|
||||
|
||||
|
||||
def test_file_md5_matches_hashlib(tmp_path: Path):
|
||||
|
|
@ -74,46 +74,3 @@ 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
|
||||
|
|
|
|||
|
|
@ -256,7 +256,6 @@ 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
|
||||
|
|
@ -401,7 +400,6 @@ 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: [])
|
||||
|
||||
|
|
|
|||
|
|
@ -285,7 +285,7 @@ def test_extra_paths_collector_records_dirs_files_notes_and_excludes(
|
|||
)
|
||||
return True
|
||||
|
||||
monkeypatch.setattr(paths.h, "stat_dir_triplet", fake_stat_triplet)
|
||||
monkeypatch.setattr(paths.h, "stat_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_dir_triplet", lambda p: ("root", "root", "0755"))
|
||||
monkeypatch.setattr(paths.h, "stat_triplet", lambda p: ("root", "root", "0755"))
|
||||
monkeypatch.setattr(
|
||||
paths, "capture_file", lambda *a, **kw: calls.append(kw["abs_path"]) or True
|
||||
)
|
||||
|
|
|
|||
|
|
@ -39,12 +39,8 @@ 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.
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue