DocForge2/src/docforge/_fs_safety.py

"""Internal directory binding helpers for disposable publication paths."""

from __future__ import annotations

import ctypes
import errno
import os
import secrets
import stat
from collections.abc import Callable
from contextlib import suppress
from pathlib import Path
from typing import Protocol, cast

from .errors import DocForgeError

RENAME_EXCHANGE = 2


class _RenameAt2(Protocol):
    argtypes: list[object]
    restype: object

    def __call__(
        self,
        old_directory_fd: int,
        old_name: bytes,
        new_directory_fd: int,
        new_name: bytes,
        flags: int,
        /,
    ) -> int: ...


def rename_exchange_at(directory_fd: int, first: str, second: str) -> None:
    """Atomically exchange two names inside one already bound directory."""

    library = ctypes.CDLL(None, use_errno=True)
    try:
        rename_at2 = cast(_RenameAt2, library.renameat2)
    except AttributeError as error:
        raise DocForgeError(
            "atomic_exchange_unavailable",
            "Atomic exchange is unavailable on this platform",
        ) from error
    rename_at2.argtypes = [
        ctypes.c_int,
        ctypes.c_char_p,
        ctypes.c_int,
        ctypes.c_char_p,
        ctypes.c_uint,
    ]
    rename_at2.restype = ctypes.c_int
    ctypes.set_errno(0)
    result = rename_at2(
        directory_fd,
        os.fsencode(first),
        directory_fd,
        os.fsencode(second),
        RENAME_EXCHANGE,
    )
    if result == 0:
        return
    error_number = ctypes.get_errno()
    if error_number in {errno.ENOSYS, errno.EINVAL, errno.EOPNOTSUPP}:
        raise DocForgeError(
            "atomic_exchange_unavailable",
            "Atomic exchange is unavailable on this filesystem",
        )
    raise DocForgeError(
        "publication_failure",
        "Could not exchange atomic publication paths",
        error_number=error_number,
    ) from OSError(error_number, os.strerror(error_number))


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)