from __future__ import annotations import json import os from pathlib import Path import stat from typing import Any from .errors import ConfigurationError _NOFOLLOW = getattr(os, "O_NOFOLLOW", 0) def reject_symlink_chain(path: Path) -> None: """Reject symlinks in every existing component of an absolute path.""" path = Path(path) if not path.is_absolute(): raise ConfigurationError(f"path_not_absolute:{path}") current = Path(path.anchor) for part in path.parts[1:]: current = current / part try: st = os.lstat(current) except FileNotFoundError: continue if stat.S_ISLNK(st.st_mode): raise ConfigurationError(f"symlink_rejected:{current}") def verify_secure_file( path: Path, *, expected_uid: int | None, allowed_write_bits: int = stat.S_IWUSR, require_executable: bool = False, ) -> os.stat_result: path = Path(path) reject_symlink_chain(path) before = os.lstat(path) if not stat.S_ISREG(before.st_mode): raise ConfigurationError(f"not_regular_file:{path}") if expected_uid is not None and before.st_uid != expected_uid: raise ConfigurationError(f"unexpected_owner:{path}") write_bits = before.st_mode & (stat.S_IWUSR | stat.S_IWGRP | stat.S_IWOTH) if write_bits & ~allowed_write_bits: raise ConfigurationError(f"permissive_mode:{path}") if require_executable and not (before.st_mode & stat.S_IXUSR): raise ConfigurationError(f"not_executable:{path}") fd = os.open(path, os.O_RDONLY | _NOFOLLOW) try: after = os.fstat(fd) if (before.st_dev, before.st_ino) != (after.st_dev, after.st_ino): raise ConfigurationError(f"inode_changed:{path}") finally: os.close(fd) return before def read_secure_bytes( path: Path, *, expected_uid: int | None, max_bytes: int = 1_048_576, ) -> bytes: path = Path(path) before = verify_secure_file(path, expected_uid=expected_uid) if before.st_size > max_bytes: raise ConfigurationError(f"file_too_large:{path}") fd = os.open(path, os.O_RDONLY | _NOFOLLOW) try: data = bytearray() while True: chunk = os.read(fd, min(65536, max_bytes + 1 - len(data))) if not chunk: break data.extend(chunk) if len(data) > max_bytes: raise ConfigurationError(f"file_too_large:{path}") return bytes(data) finally: os.close(fd) def read_secure_text(path: Path, *, expected_uid: int | None, max_bytes: int = 1_048_576) -> str: try: return read_secure_bytes(path, expected_uid=expected_uid, max_bytes=max_bytes).decode("utf-8").strip() except UnicodeDecodeError as exc: raise ConfigurationError(f"not_utf8:{path}") from exc def read_secure_json(path: Path, *, expected_uid: int | None, max_bytes: int = 1_048_576) -> dict[str, Any]: try: value = json.loads(read_secure_text(path, expected_uid=expected_uid, max_bytes=max_bytes)) except json.JSONDecodeError as exc: raise ConfigurationError(f"invalid_json:{path}") from exc if not isinstance(value, dict): raise ConfigurationError(f"json_root_not_object:{path}") return value def ensure_secure_directory(path: Path, *, mode: int = 0o700) -> None: path = Path(path) if not path.is_absolute(): raise ConfigurationError(f"path_not_absolute:{path}") reject_symlink_chain(path.parent) path.mkdir(parents=True, exist_ok=True, mode=mode) reject_symlink_chain(path) st = os.lstat(path) if not stat.S_ISDIR(st.st_mode): raise ConfigurationError(f"not_directory:{path}") if st.st_mode & (stat.S_IWGRP | stat.S_IWOTH): raise ConfigurationError(f"permissive_directory:{path}") def atomic_write_bytes(path: Path, payload: bytes, *, mode: int = 0o600) -> None: """Write through a verified directory fd and atomically replace the target.""" path = Path(path) ensure_secure_directory(path.parent) parent_before = os.lstat(path.parent) parent_fd = os.open(path.parent, os.O_RDONLY | os.O_DIRECTORY | _NOFOLLOW) temp_name = f".{path.name}.tmp-{os.getpid()}-{os.urandom(6).hex()}" try: parent_after = os.fstat(parent_fd) if (parent_before.st_dev, parent_before.st_ino) != ( parent_after.st_dev, parent_after.st_ino, ): raise ConfigurationError(f"directory_changed:{path.parent}") fd = os.open( temp_name, os.O_WRONLY | os.O_CREAT | os.O_EXCL | _NOFOLLOW, mode, dir_fd=parent_fd, ) try: view = memoryview(payload) while view: written = os.write(fd, view) view = view[written:] os.fsync(fd) finally: os.close(fd) os.replace( temp_name, path.name, src_dir_fd=parent_fd, dst_dir_fd=parent_fd, ) os.fsync(parent_fd) finally: try: os.unlink(temp_name, dir_fd=parent_fd) except FileNotFoundError: pass os.close(parent_fd) def atomic_write_json(path: Path, value: Any, *, mode: int = 0o600) -> None: payload = (json.dumps(value, indent=2, sort_keys=True) + "\n").encode("utf-8") atomic_write_bytes(path, payload, mode=mode) def ensure_shared_directory(path: Path, *, group_writable: bool = False) -> None: """Verify/create a setgid queue directory shared with a consumer group.""" path = Path(path) if not path.is_absolute(): raise ConfigurationError(f"path_not_absolute:{path}") reject_symlink_chain(path.parent) mode = 0o2770 if group_writable else 0o2750 try: os.lstat(path) except FileNotFoundError: path.mkdir(parents=True, mode=mode) os.chmod(path, mode) reject_symlink_chain(path) st = os.lstat(path) if not stat.S_ISDIR(st.st_mode): raise ConfigurationError(f"not_directory:{path}") observed = stat.S_IMODE(st.st_mode) required = stat.S_ISGID | stat.S_IRGRP | stat.S_IXGRP if group_writable: required |= stat.S_IWGRP if observed & 0o007 or observed & required != required: raise ConfigurationError(f"invalid_shared_directory_mode:{path}") def atomic_write_bytes_shared( path: Path, payload: bytes, *, mode: int = 0o640, group_writable_directory: bool = False, ) -> None: """Atomic write into a setgid/group-readable shared queue directory.""" path = Path(path) ensure_shared_directory(path.parent, group_writable=group_writable_directory) parent_before = os.lstat(path.parent) parent_fd = os.open(path.parent, os.O_RDONLY | os.O_DIRECTORY | _NOFOLLOW) temp_name = f".{path.name}.tmp-{os.getpid()}-{os.urandom(6).hex()}" try: parent_after = os.fstat(parent_fd) if (parent_before.st_dev, parent_before.st_ino) != ( parent_after.st_dev, parent_after.st_ino, ): raise ConfigurationError(f"directory_changed:{path.parent}") fd = os.open( temp_name, os.O_WRONLY | os.O_CREAT | os.O_EXCL | _NOFOLLOW, mode, dir_fd=parent_fd, ) os.fchmod(fd, mode) try: view = memoryview(payload) while view: written = os.write(fd, view) view = view[written:] os.fsync(fd) finally: os.close(fd) os.replace(temp_name, path.name, src_dir_fd=parent_fd, dst_dir_fd=parent_fd) os.fsync(parent_fd) finally: try: os.unlink(temp_name, dir_fd=parent_fd) except FileNotFoundError: pass os.close(parent_fd)