Compare commits
1 commit
bc4af135b4
...
87ee83d3ee
| Author | SHA1 | Date | |
|---|---|---|---|
| 87ee83d3ee |
2 changed files with 45 additions and 137 deletions
|
|
@ -5,9 +5,8 @@ from __future__ import annotations
|
||||||
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
|
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,
|
||||||
refuse symlink parents, reject root-run output through untrusted parent
|
write through a private temporary file in the target directory, and replace the
|
||||||
components, write through a temporary file in the target directory, and replace
|
final path atomically. Existing final-path symlinks are refused rather than
|
||||||
the final path atomically. Existing final-path symlinks are refused rather than
|
|
||||||
followed.
|
followed.
|
||||||
"""
|
"""
|
||||||
|
|
||||||
|
|
@ -21,111 +20,8 @@ class OutputPathError(OSError):
|
||||||
"""Raised when a requested output path is unsafe."""
|
"""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:
|
def _absolute(path: Path) -> Path:
|
||||||
expanded = path.expanduser()
|
return path if path.is_absolute() else Path.cwd() / path
|
||||||
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:
|
def _check_existing_output_file(path: Path) -> None:
|
||||||
|
|
@ -140,36 +36,65 @@ def _check_existing_output_file(path: Path) -> None:
|
||||||
|
|
||||||
|
|
||||||
def _check_parent_components(parent: Path) -> None:
|
def _check_parent_components(parent: Path) -> None:
|
||||||
"""Require every existing parent component to be a trusted real directory."""
|
"""Require every existing parent component to be a real directory."""
|
||||||
|
|
||||||
_mkdir_safe_dir_tree(parent)
|
parent = _absolute(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:
|
def ensure_safe_directory(path: Path) -> None:
|
||||||
"""Create or validate a directory tree without accepting unsafe parents."""
|
"""Create or validate a directory tree without accepting symlinks."""
|
||||||
|
|
||||||
_mkdir_safe_dir_tree(path)
|
path = _absolute(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:
|
def write_text_safely(path: Path, text: str, *, encoding: str = "utf-8") -> None:
|
||||||
"""Write text without following a final-path symlink.
|
"""Write text without following a final-path symlink.
|
||||||
|
|
||||||
The target's parent is created/validated component by component. Existing
|
The target's parent must already exist and every parent component must be a
|
||||||
parent symlinks are refused for every user. When running as root, existing
|
real directory. The write is completed with os.replace(), which atomically
|
||||||
parent components must also be root-owned and not writable by group/other,
|
swaps the final directory entry and does not dereference a final symlink.
|
||||||
except for sticky shared boundaries such as /tmp. Existing final-path
|
|
||||||
symlinks are refused rather than followed.
|
|
||||||
"""
|
"""
|
||||||
|
|
||||||
path = _absolute(path)
|
path = _absolute(path)
|
||||||
parent = _mkdir_safe_dir_tree(path.parent)
|
_check_parent_components(path.parent)
|
||||||
_check_existing_output_file(path)
|
_check_existing_output_file(path)
|
||||||
|
|
||||||
fd = -1
|
fd = -1
|
||||||
tmp_name: str | None = None
|
tmp_name: str | None = None
|
||||||
try:
|
try:
|
||||||
fd, tmp_name = tempfile.mkstemp(
|
fd, tmp_name = tempfile.mkstemp(
|
||||||
prefix=f".{path.name}.", suffix=".tmp", dir=str(parent), text=True
|
prefix=f".{path.name}.", suffix=".tmp", dir=str(path.parent), text=True
|
||||||
)
|
)
|
||||||
with os.fdopen(fd, "w", encoding=encoding) as f:
|
with os.fdopen(fd, "w", encoding=encoding) as f:
|
||||||
fd = -1
|
fd = -1
|
||||||
|
|
@ -177,7 +102,6 @@ def write_text_safely(path: Path, text: str, *, encoding: str = "utf-8") -> None
|
||||||
f.flush()
|
f.flush()
|
||||||
os.fsync(f.fileno())
|
os.fsync(f.fileno())
|
||||||
os.chmod(tmp_name, 0o600)
|
os.chmod(tmp_name, 0o600)
|
||||||
_check_parent_components(path.parent)
|
|
||||||
_check_existing_output_file(path)
|
_check_existing_output_file(path)
|
||||||
os.replace(tmp_name, path)
|
os.replace(tmp_name, path)
|
||||||
tmp_name = None
|
tmp_name = None
|
||||||
|
|
|
||||||
|
|
@ -118,19 +118,3 @@ def test_cli_reports_unsafe_output_path_without_overwriting_symlink(tmp_path: Pa
|
||||||
|
|
||||||
assert exit_code == 2
|
assert exit_code == 2
|
||||||
assert target.read_text(encoding="utf-8") == "keep\n"
|
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()
|
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue