248 lines
8.4 KiB
Python
248 lines
8.4 KiB
Python
"""Internal directory binding helpers for disposable publication paths."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import secrets
|
|
import stat
|
|
from collections.abc import Callable
|
|
from contextlib import suppress
|
|
from pathlib import Path
|
|
|
|
from .errors import DocForgeError
|
|
|
|
|
|
def open_bound_directory(path: Path) -> int:
|
|
"""Open one real directory and bind its current inode for later operations."""
|
|
|
|
try:
|
|
path_status = path.lstat()
|
|
if (
|
|
stat.S_ISLNK(path_status.st_mode)
|
|
or not stat.S_ISDIR(path_status.st_mode)
|
|
or path.resolve(strict=True) != path
|
|
):
|
|
raise DocForgeError(
|
|
"path_escape",
|
|
"Derived cache root is not a safe real directory",
|
|
)
|
|
directory_fd = os.open(path, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW)
|
|
except FileNotFoundError as error:
|
|
raise DocForgeError(
|
|
"missing_index",
|
|
"Derived cache root does not exist",
|
|
) from error
|
|
except OSError as error:
|
|
raise DocForgeError(
|
|
"path_escape",
|
|
"Derived cache root cannot be opened safely",
|
|
) from error
|
|
try:
|
|
opened_status = os.fstat(directory_fd)
|
|
if opened_status.st_dev != path_status.st_dev or opened_status.st_ino != path_status.st_ino:
|
|
raise DocForgeError(
|
|
"path_escape",
|
|
"Derived cache root changed while opening",
|
|
)
|
|
except Exception:
|
|
os.close(directory_fd)
|
|
raise
|
|
return directory_fd
|
|
|
|
|
|
def require_bound_directory(path: Path, directory_fd: int) -> None:
|
|
"""Require a path to still name the exact opened real directory."""
|
|
|
|
try:
|
|
path_status = path.lstat()
|
|
opened_status = os.fstat(directory_fd)
|
|
if (
|
|
stat.S_ISLNK(path_status.st_mode)
|
|
or not stat.S_ISDIR(path_status.st_mode)
|
|
or path.resolve(strict=True) != path
|
|
or opened_status.st_dev != path_status.st_dev
|
|
or opened_status.st_ino != path_status.st_ino
|
|
):
|
|
raise DocForgeError(
|
|
"path_escape",
|
|
"Derived cache root changed during publication",
|
|
)
|
|
except FileNotFoundError as error:
|
|
raise DocForgeError(
|
|
"path_escape",
|
|
"Derived cache root disappeared during publication",
|
|
) from error
|
|
|
|
|
|
def open_confined_directory(root: Path, path: Path, *, create: bool) -> int:
|
|
"""Open a descendant directory through stable no-follow directory descriptors."""
|
|
|
|
try:
|
|
unsafe = (
|
|
root.is_symlink()
|
|
or root.resolve(strict=True) != root
|
|
or not path.is_relative_to(root)
|
|
or path == root
|
|
)
|
|
except OSError as error:
|
|
raise DocForgeError("path_escape", "Project root cannot be resolved safely") from error
|
|
if unsafe:
|
|
raise DocForgeError("path_escape", "Derived output directory is not confined")
|
|
relative = path.relative_to(root)
|
|
try:
|
|
descriptor = os.open(root, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW)
|
|
except OSError as error:
|
|
raise DocForgeError("path_escape", "Project root cannot be opened safely") from error
|
|
try:
|
|
for part in relative.parts:
|
|
if part in {"", ".", ".."}:
|
|
raise DocForgeError("path_escape", "Derived output directory is not confined")
|
|
if create:
|
|
try:
|
|
os.mkdir(part, mode=0o700, dir_fd=descriptor)
|
|
except FileExistsError:
|
|
pass
|
|
except OSError as error:
|
|
raise DocForgeError(
|
|
"publication_failure",
|
|
"Derived output directory could not be created",
|
|
) from error
|
|
try:
|
|
next_descriptor = os.open(
|
|
part,
|
|
os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW,
|
|
dir_fd=descriptor,
|
|
)
|
|
except OSError as error:
|
|
raise DocForgeError(
|
|
"path_escape",
|
|
"Derived output directory is missing or unsafe",
|
|
) from error
|
|
os.close(descriptor)
|
|
descriptor = next_descriptor
|
|
require_bound_directory(path, descriptor)
|
|
return descriptor
|
|
except Exception:
|
|
os.close(descriptor)
|
|
raise
|
|
|
|
|
|
def safe_file_identity_at(
|
|
directory: Path,
|
|
directory_fd: int,
|
|
name: str,
|
|
) -> dict[str, object] | None:
|
|
"""Return one no-follow regular-file identity relative to a bound directory."""
|
|
|
|
del directory
|
|
if not name or "/" in name or name in {".", ".."}:
|
|
raise DocForgeError("path_escape", "Derived artifact name is unsafe")
|
|
try:
|
|
current = os.stat(name, dir_fd=directory_fd, follow_symlinks=False)
|
|
except FileNotFoundError:
|
|
return None
|
|
except OSError as error:
|
|
raise DocForgeError("path_escape", "Derived artifact cannot be inspected") from error
|
|
if not stat.S_ISREG(current.st_mode):
|
|
raise DocForgeError("path_escape", "Derived artifact is not a safe regular file")
|
|
return {
|
|
"path": name,
|
|
"device": current.st_dev,
|
|
"inode": current.st_ino,
|
|
"mode": current.st_mode,
|
|
"size": current.st_size,
|
|
"mtime_ns": current.st_mtime_ns,
|
|
"ctime_ns": current.st_ctime_ns,
|
|
}
|
|
|
|
|
|
def read_bounded_file_at(
|
|
directory_fd: int,
|
|
name: str,
|
|
maximum_bytes: int,
|
|
) -> bytes | None:
|
|
"""Read one regular file through a bound directory without following links."""
|
|
|
|
try:
|
|
descriptor = os.open(name, os.O_RDONLY | os.O_NOFOLLOW, dir_fd=directory_fd)
|
|
except FileNotFoundError:
|
|
return None
|
|
except OSError as error:
|
|
raise DocForgeError("path_escape", "Derived artifact cannot be opened safely") from error
|
|
with os.fdopen(descriptor, "rb") as handle:
|
|
current = os.fstat(handle.fileno())
|
|
if not stat.S_ISREG(current.st_mode) or current.st_size > maximum_bytes:
|
|
raise DocForgeError("invalid_projection", "Derived artifact is invalid or oversized")
|
|
content = handle.read(maximum_bytes + 1)
|
|
if len(content) > maximum_bytes:
|
|
raise DocForgeError("invalid_projection", "Derived artifact is oversized")
|
|
return content
|
|
|
|
|
|
def atomic_replace_bytes_at(
|
|
path: Path,
|
|
directory_fd: int,
|
|
name: str,
|
|
content: bytes,
|
|
*,
|
|
verify: Callable[[], None],
|
|
) -> dict[str, object]:
|
|
"""Durably replace one file inside an already bound directory."""
|
|
|
|
if not name or "/" in name or name in {".", ".."}:
|
|
raise DocForgeError("path_escape", "Derived artifact name is unsafe")
|
|
existing = safe_file_identity_at(path, directory_fd, name)
|
|
del existing
|
|
temporary = f".docforge-projection-{secrets.token_hex(12)}"
|
|
descriptor: int | None = None
|
|
committed = False
|
|
try:
|
|
descriptor = os.open(
|
|
temporary,
|
|
os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW,
|
|
0o600,
|
|
dir_fd=directory_fd,
|
|
)
|
|
with os.fdopen(descriptor, "wb") as handle:
|
|
descriptor = None
|
|
handle.write(content)
|
|
handle.flush()
|
|
os.fsync(handle.fileno())
|
|
verify()
|
|
require_bound_directory(path, directory_fd)
|
|
os.replace(
|
|
temporary,
|
|
name,
|
|
src_dir_fd=directory_fd,
|
|
dst_dir_fd=directory_fd,
|
|
)
|
|
committed = True
|
|
os.fsync(directory_fd)
|
|
identity = safe_file_identity_at(path, directory_fd, name)
|
|
if identity is None:
|
|
raise DocForgeError(
|
|
"publication_failure",
|
|
"Derived artifact disappeared after publication",
|
|
mutation_committed=True,
|
|
)
|
|
return identity
|
|
except DocForgeError as error:
|
|
if committed:
|
|
raise DocForgeError(
|
|
"publication_failure",
|
|
"Derived artifact was replaced but final publication verification failed",
|
|
mutation_committed=True,
|
|
cause=error.code,
|
|
) from error
|
|
raise
|
|
except OSError as error:
|
|
raise DocForgeError(
|
|
"publication_failure",
|
|
"Derived artifact publication failed",
|
|
mutation_committed=committed,
|
|
) from error
|
|
finally:
|
|
if descriptor is not None:
|
|
os.close(descriptor)
|
|
with suppress(OSError):
|
|
os.unlink(temporary, dir_fd=directory_fd)
|