From a901c9705bf728b3d8e648007985fe23d82769d3 Mon Sep 17 00:00:00 2001 From: Andraxion Date: Wed, 29 Jul 2026 16:03:14 -0400 Subject: [PATCH] Harden canonical publication against races --- src/docforge/_fs_safety.py | 62 +++ src/docforge/application.py | 648 +++++++++++++++++++++++++++++--- src/docforge/changesets.py | 10 + src/docforge/incremental.py | 29 +- src/docforge/rendering.py | 36 +- tests/test_changesets.py | 288 ++++++++++++++ tests/test_incremental_cache.py | 20 + tests/test_rendering.py | 44 +++ 8 files changed, 1059 insertions(+), 78 deletions(-) diff --git a/src/docforge/_fs_safety.py b/src/docforge/_fs_safety.py index 6795a77..81c9f3b 100644 --- a/src/docforge/_fs_safety.py +++ b/src/docforge/_fs_safety.py @@ -2,15 +2,77 @@ 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.""" diff --git a/src/docforge/application.py b/src/docforge/application.py index 3efb2b1..b9292da 100644 --- a/src/docforge/application.py +++ b/src/docforge/application.py @@ -4,13 +4,16 @@ from __future__ import annotations import json import os +import stat import tempfile from collections import defaultdict from collections.abc import Mapping, Sequence from contextlib import suppress +from dataclasses import dataclass from pathlib import Path -from typing import Protocol, cast +from typing import Literal, Protocol, cast +from ._fs_safety import open_bound_directory, rename_exchange_at, require_bound_directory from .changesets import ChangesetStore from .errors import DocForgeError from .index import ProjectIndex @@ -30,6 +33,58 @@ class CanonicalApplier(Protocol): ) -> dict[str, object]: ... +@dataclass(frozen=True) +class _CanonicalFile: + device: int + inode: int + mode: int + owner: int + group: int + size: int + modified_ns: int + changed_ns: int + content: bytes + + def unchanged(self, other: _CanonicalFile | None) -> bool: + return self == other + + def renamed_to(self, other: _CanonicalFile | None) -> bool: + if other is None: + return False + return ( + self.device, + self.inode, + self.mode, + self.owner, + self.group, + self.size, + self.modified_ns, + self.content, + ) == ( + other.device, + other.inode, + other.mode, + other.owner, + other.group, + other.size, + other.modified_ns, + other.content, + ) + + +@dataclass +class _CanonicalPublication: + relative: str + target: Path + action: Literal["create", "update", "delete"] + expected: _CanonicalFile | None + staged: Path + staged_snapshot: _CanonicalFile + backup_snapshot: _CanonicalFile | None = None + published_snapshot: _CanonicalFile | None = None + committed: bool = False + + class GenericCanonicalApplier: """Apply generic Markdown/TOML projections inside declared content roots.""" @@ -46,35 +101,49 @@ class GenericCanonicalApplier: changed_sources = self._changed_sources(base, projected) if not changed_sources: raise DocForgeError("empty_changeset", "Changeset produces no canonical changes") + base_by_source = self._nodes_by_source(base) projected_by_source = self._nodes_by_source(projected) - staged: dict[Path, Path] = {} - previous: dict[Path, bytes | None] = {} + publications: list[_CanonicalPublication] = [] created_directories: list[Path] = [] targets = {relative: self._target(relative) for relative in sorted(changed_sources)} try: for relative, target in targets.items(): - previous[target] = target.read_bytes() if target.is_file() else None - nodes = projected_by_source.get(relative, ()) - if not nodes: - continue self._prepare_parent(target.parent, created_directories) - raw = self._serialize_source(projected, relative, nodes) - if len(raw) > base.descriptor.limits.max_source_bytes: + expected = self._capture(target, base.descriptor.limits.max_source_bytes) + existed_in_base = relative in base_by_source + if existed_in_base != (expected is not None): + raise DocForgeError( + "base_conflict", + "Canonical target changed before application staging", + source=relative, + ) + nodes = projected_by_source.get(relative, ()) + raw = self._serialize_source(projected, relative, nodes) if nodes else b"" + if nodes and len(raw) > base.descriptor.limits.max_source_bytes: raise DocForgeError( "source_too_large", "Applied canonical source exceeds the configured limit", source=relative, ) - descriptor, temporary_name = tempfile.mkstemp( - prefix=".docforge-apply-", - dir=target.parent, + staged, staged_snapshot = self._stage( + target, + raw, + mode=(stat.S_IMODE(expected.mode) if expected is not None else 0o600), + ) + publications.append( + _CanonicalPublication( + relative=relative, + target=target, + action=( + "delete" + if not nodes + else ("update" if expected is not None else "create") + ), + expected=expected, + staged=staged, + staged_snapshot=staged_snapshot, + ) ) - temporary = Path(temporary_name) - with os.fdopen(descriptor, "wb") as handle: - handle.write(raw) - handle.flush() - os.fsync(handle.fileno()) - staged[target] = temporary current = self.project.load() if current.source_hash != base.source_hash or current.revision != base.revision: @@ -89,13 +158,8 @@ class GenericCanonicalApplier: for target in targets.values(): if target.is_symlink(): raise DocForgeError("path_escape", "Canonical target became a symbolic link") - for target in sorted(targets.values(), key=str): - temporary = staged.get(target) - if temporary is None: - target.unlink(missing_ok=True) - else: - os.replace(temporary, target) - self._fsync_directory(target.parent) + for publication in publications: + self._publish(publication) applied = self.project.load() if self._semantic_snapshot(applied) != self._semantic_snapshot(projected): @@ -103,19 +167,522 @@ class GenericCanonicalApplier: "application_mismatch", "Applied canonical files do not reproduce the validated proposal", ) - except Exception: - for temporary in staged.values(): - temporary.unlink(missing_ok=True) - self._restore(previous) + except Exception as error: + recovery = self._rollback(publications) + self._discard_unowned_staging(publications) self._remove_empty_directories(created_directories) + if recovery: + cause = error.code if isinstance(error, DocForgeError) else type(error).__name__ + raise DocForgeError( + "application_recovery_required", + ( + "Canonical application raced external writes and could not be " + "rolled back without overwriting foreign data" + ), + cause=cause, + conflicts=recovery, + remediation=( + "Preserve the retained files, compare them with the named canonical " + "targets, and resolve the project before retrying a new changeset." + ), + ) from error + if isinstance(error, DocForgeError) and error.code == "application_recovery_required": + raise DocForgeError( + "base_conflict", + "Canonical target raced publication but was restored without data loss", + ) from error raise + retained = self._discard_backups(publications) return { "applied_sources": sorted(changed_sources), "removed_sources": sorted( source for source in changed_sources if source not in projected_by_source ), + "retained_recovery_files": retained, } + def _publish(self, publication: _CanonicalPublication) -> None: + target = publication.target + parent_fd = open_bound_directory(target.parent) + try: + current = self._capture_at( + target.parent, + parent_fd, + target.name, + self.project.descriptor.limits.max_source_bytes, + ) + if publication.expected is None: + if current is not None: + raise DocForgeError( + "base_conflict", + "Canonical create target appeared during application", + source=publication.relative, + ) + elif not publication.expected.unchanged(current): + raise DocForgeError( + "base_conflict", + "Canonical target changed during application", + source=publication.relative, + ) + staged = self._capture_at( + target.parent, + parent_fd, + publication.staged.name, + self.project.descriptor.limits.max_source_bytes, + ) + if staged is None or not publication.staged_snapshot.unchanged(staged): + raise DocForgeError( + "application_mismatch", + "Canonical staging file changed before publication", + source=publication.relative, + ) + if publication.action == "create": + self._publish_create(publication, parent_fd) + else: + self._publish_exchange(publication, parent_fd) + finally: + os.close(parent_fd) + + def _publish_create( + self, + publication: _CanonicalPublication, + parent_fd: int, + ) -> None: + try: + os.link( + publication.staged.name, + publication.target.name, + src_dir_fd=parent_fd, + dst_dir_fd=parent_fd, + follow_symlinks=False, + ) + except FileExistsError as error: + raise DocForgeError( + "base_conflict", + "Canonical create target appeared during application", + source=publication.relative, + ) from error + except OSError as error: + raise DocForgeError( + "application_failure", + "Canonical create target could not be published", + source=publication.relative, + ) from error + publication.committed = True + published = self._capture_at( + publication.target.parent, + parent_fd, + publication.target.name, + self.project.descriptor.limits.max_source_bytes, + ) + if not publication.staged_snapshot.renamed_to(published): + raise DocForgeError( + "application_mismatch", + "Published canonical create target does not match its staging file", + source=publication.relative, + ) + publication.published_snapshot = published + os.unlink(publication.staged.name, dir_fd=parent_fd) + os.fsync(parent_fd) + require_bound_directory(publication.target.parent, parent_fd) + + def _publish_exchange( + self, + publication: _CanonicalPublication, + parent_fd: int, + ) -> None: + rename_exchange_at(parent_fd, publication.staged.name, publication.target.name) + publication.committed = True + displaced = self._capture_at( + publication.target.parent, + parent_fd, + publication.staged.name, + self.project.descriptor.limits.max_source_bytes, + ) + published = self._capture_at( + publication.target.parent, + parent_fd, + publication.target.name, + self.project.descriptor.limits.max_source_bytes, + ) + publication.backup_snapshot = displaced + publication.published_snapshot = published + if ( + publication.expected is None + or not publication.expected.renamed_to(displaced) + or not publication.staged_snapshot.renamed_to(published) + ): + if not self._exchange_back(publication, parent_fd): + raise DocForgeError( + "application_recovery_required", + "Canonical target raced atomic publication and displaced data was retained", + source=publication.relative, + retained=self._relative(publication.staged), + ) + raise DocForgeError( + "base_conflict", + "Canonical target changed at the atomic publication boundary", + source=publication.relative, + ) + if publication.action == "delete": + os.unlink(publication.target.name, dir_fd=parent_fd) + publication.published_snapshot = None + os.fsync(parent_fd) + require_bound_directory(publication.target.parent, parent_fd) + + def _exchange_back( + self, + publication: _CanonicalPublication, + parent_fd: int, + ) -> bool: + backup = publication.backup_snapshot + published = publication.published_snapshot + if backup is None or published is None: + return False + try: + rename_exchange_at(parent_fd, publication.staged.name, publication.target.name) + restored = self._capture_at( + publication.target.parent, + parent_fd, + publication.target.name, + self.project.descriptor.limits.max_source_bytes, + ) + staged = self._capture_at( + publication.target.parent, + parent_fd, + publication.staged.name, + self.project.descriptor.limits.max_source_bytes, + ) + if not backup.renamed_to(restored) or not published.renamed_to(staged): + return False + publication.backup_snapshot = None + publication.published_snapshot = None + publication.committed = False + os.fsync(parent_fd) + return True + except (DocForgeError, OSError): + return False + + def _rollback( + self, + publications: Sequence[_CanonicalPublication], + ) -> list[dict[str, object]]: + conflicts: list[dict[str, object]] = [] + for publication in reversed(publications): + if not publication.committed: + continue + try: + if publication.action == "create": + conflict = self._rollback_create(publication) + elif publication.action == "delete" and publication.published_snapshot is None: + conflict = self._rollback_delete(publication) + else: + conflict = self._rollback_update(publication) + except Exception as error: + conflict = self._recovery_conflict(publication, "rollback_failed") + conflict["error"] = ( + error.code if isinstance(error, DocForgeError) else type(error).__name__ + ) + if conflict is not None: + conflicts.append(conflict) + return conflicts + + def _rollback_update( + self, + publication: _CanonicalPublication, + ) -> dict[str, object] | None: + parent_fd = open_bound_directory(publication.target.parent) + try: + current = self._capture_at( + publication.target.parent, + parent_fd, + publication.target.name, + self.project.descriptor.limits.max_source_bytes, + ) + backup = self._capture_at( + publication.target.parent, + parent_fd, + publication.staged.name, + self.project.descriptor.limits.max_source_bytes, + ) + if ( + publication.published_snapshot is None + or publication.backup_snapshot is None + or not publication.published_snapshot.unchanged(current) + or not publication.backup_snapshot.unchanged(backup) + ): + return self._recovery_conflict(publication, "target_or_backup_changed") + if not self._exchange_back(publication, parent_fd): + return self._recovery_conflict(publication, "atomic_restore_unconfirmed") + os.unlink(publication.staged.name, dir_fd=parent_fd) + os.fsync(parent_fd) + return None + finally: + os.close(parent_fd) + + def _rollback_delete( + self, + publication: _CanonicalPublication, + ) -> dict[str, object] | None: + parent_fd = open_bound_directory(publication.target.parent) + try: + current = self._capture_at( + publication.target.parent, + parent_fd, + publication.target.name, + self.project.descriptor.limits.max_source_bytes, + ) + backup = self._capture_at( + publication.target.parent, + parent_fd, + publication.staged.name, + self.project.descriptor.limits.max_source_bytes, + ) + if current is not None: + return self._recovery_conflict(publication, "deleted_target_reappeared") + if publication.backup_snapshot is None or not publication.backup_snapshot.unchanged( + backup + ): + return self._recovery_conflict(publication, "backup_changed") + try: + os.link( + publication.staged.name, + publication.target.name, + src_dir_fd=parent_fd, + dst_dir_fd=parent_fd, + follow_symlinks=False, + ) + except FileExistsError: + return self._recovery_conflict(publication, "deleted_target_reappeared") + restored = self._capture_at( + publication.target.parent, + parent_fd, + publication.target.name, + self.project.descriptor.limits.max_source_bytes, + ) + if not publication.backup_snapshot.renamed_to(restored): + return self._recovery_conflict(publication, "restore_unconfirmed") + os.unlink(publication.staged.name, dir_fd=parent_fd) + publication.backup_snapshot = None + publication.committed = False + os.fsync(parent_fd) + return None + finally: + os.close(parent_fd) + + def _rollback_create( + self, + publication: _CanonicalPublication, + ) -> dict[str, object] | None: + parent_fd = open_bound_directory(publication.target.parent) + tombstone: Path | None = None + try: + current = self._capture_at( + publication.target.parent, + parent_fd, + publication.target.name, + self.project.descriptor.limits.max_source_bytes, + ) + if current is None: + publication.committed = False + return None + if ( + publication.published_snapshot is None + or not publication.published_snapshot.unchanged(current) + ): + return self._recovery_conflict(publication, "created_target_changed") + tombstone, tombstone_snapshot = self._stage(publication.target, b"", mode=0o600) + rename_exchange_at(parent_fd, tombstone.name, publication.target.name) + displaced = self._capture_at( + publication.target.parent, + parent_fd, + tombstone.name, + self.project.descriptor.limits.max_source_bytes, + ) + published_tombstone = self._capture_at( + publication.target.parent, + parent_fd, + publication.target.name, + self.project.descriptor.limits.max_source_bytes, + ) + if not publication.published_snapshot.renamed_to( + displaced + ) or not tombstone_snapshot.renamed_to(published_tombstone): + with suppress(DocForgeError): + rename_exchange_at(parent_fd, tombstone.name, publication.target.name) + return self._recovery_conflict(publication, "create_rollback_raced") + os.unlink(publication.target.name, dir_fd=parent_fd) + os.unlink(tombstone.name, dir_fd=parent_fd) + tombstone = None + publication.committed = False + publication.published_snapshot = None + os.fsync(parent_fd) + return None + finally: + if tombstone is not None: + tombstone.unlink(missing_ok=True) + os.close(parent_fd) + + def _discard_backups( + self, + publications: Sequence[_CanonicalPublication], + ) -> list[dict[str, object]]: + retained: list[dict[str, object]] = [] + for publication in publications: + if publication.action != "create" and publication.staged.exists(): + parent_fd = open_bound_directory(publication.target.parent) + try: + backup = self._capture_at( + publication.target.parent, + parent_fd, + publication.staged.name, + self.project.descriptor.limits.max_source_bytes, + ) + if ( + publication.backup_snapshot is None + or not publication.backup_snapshot.unchanged(backup) + ): + retained.append( + self._recovery_conflict(publication, "backup_cleanup_raced") + ) + continue + os.unlink(publication.staged.name, dir_fd=parent_fd) + os.fsync(parent_fd) + finally: + os.close(parent_fd) + publication.committed = False + publication.backup_snapshot = None + return retained + + @staticmethod + def _discard_unowned_staging(publications: Sequence[_CanonicalPublication]) -> None: + for publication in publications: + if not publication.committed and publication.staged.exists(): + publication.staged.unlink() + + def _stage(self, target: Path, content: bytes, *, mode: int) -> tuple[Path, _CanonicalFile]: + descriptor, temporary_name = tempfile.mkstemp( + prefix=".docforge-apply-", + dir=target.parent, + ) + temporary = Path(temporary_name) + try: + os.fchmod(descriptor, mode) + with os.fdopen(descriptor, "wb") as handle: + handle.write(content) + handle.flush() + os.fsync(handle.fileno()) + snapshot = self._capture(temporary, max(1, len(content))) + if snapshot is None: + raise DocForgeError( + "application_mismatch", + "Canonical staging file disappeared", + source=self._relative(target), + ) + return temporary, snapshot + except Exception: + with suppress(OSError): + os.close(descriptor) + temporary.unlink(missing_ok=True) + raise + + def _capture(self, path: Path, maximum: int) -> _CanonicalFile | None: + parent_fd = open_bound_directory(path.parent) + try: + return self._capture_at(path.parent, parent_fd, path.name, maximum) + finally: + os.close(parent_fd) + + @staticmethod + def _capture_at( + parent: Path, + parent_fd: int, + name: str, + maximum: int, + ) -> _CanonicalFile | None: + del parent + try: + descriptor = os.open(name, os.O_RDONLY | os.O_NOFOLLOW, dir_fd=parent_fd) + except FileNotFoundError: + return None + except OSError as error: + raise DocForgeError( + "path_escape", + "Canonical target cannot be opened safely", + target=name, + ) from error + with os.fdopen(descriptor, "rb") as handle: + before = os.fstat(handle.fileno()) + if not stat.S_ISREG(before.st_mode) or before.st_size > maximum: + raise DocForgeError( + "source_too_large", + "Canonical target is invalid or exceeds its configured limit", + target=name, + ) + content = handle.read(maximum + 1) + after = os.fstat(handle.fileno()) + if len(content) > maximum: + raise DocForgeError( + "source_too_large", + "Canonical target exceeds its configured limit", + target=name, + ) + current = os.stat(name, dir_fd=parent_fd, follow_symlinks=False) + before_identity = ( + before.st_dev, + before.st_ino, + before.st_mode, + before.st_uid, + before.st_gid, + before.st_size, + before.st_mtime_ns, + before.st_ctime_ns, + ) + after_identity = ( + after.st_dev, + after.st_ino, + after.st_mode, + after.st_uid, + after.st_gid, + after.st_size, + after.st_mtime_ns, + after.st_ctime_ns, + ) + current_identity = ( + current.st_dev, + current.st_ino, + current.st_mode, + current.st_uid, + current.st_gid, + current.st_size, + current.st_mtime_ns, + current.st_ctime_ns, + ) + if before_identity != after_identity or before_identity != current_identity: + raise DocForgeError( + "base_conflict", + "Canonical target changed while it was captured", + target=name, + ) + return _CanonicalFile(*before_identity, content) + + def _recovery_conflict( + self, + publication: _CanonicalPublication, + reason: str, + ) -> dict[str, object]: + return { + "source": publication.relative, + "reason": reason, + "target": self._relative(publication.target), + "retained": ( + self._relative(publication.staged) if publication.staged.exists() else None + ), + } + + def _relative(self, path: Path) -> str: + return path.relative_to(self.project.descriptor.root).as_posix() + def _target(self, relative: str) -> Path: candidate = Path(relative) if candidate.is_absolute() or ".." in candidate.parts: @@ -162,29 +729,6 @@ class GenericCanonicalApplier: finally: os.close(descriptor) - def _restore(self, previous: dict[Path, bytes | None]) -> None: - for target in sorted(previous, key=str): - raw = previous[target] - if raw is None: - target.unlink(missing_ok=True) - continue - target.parent.mkdir(parents=True, exist_ok=True) - descriptor, temporary_name = tempfile.mkstemp( - prefix=".docforge-rollback-", - dir=target.parent, - ) - temporary = Path(temporary_name) - try: - with os.fdopen(descriptor, "wb") as handle: - handle.write(raw) - handle.flush() - os.fsync(handle.fileno()) - os.replace(temporary, target) - self._fsync_directory(target.parent) - except Exception: - temporary.unlink(missing_ok=True) - raise - @staticmethod def _remove_empty_directories(paths: Sequence[Path]) -> None: for path in reversed(paths): diff --git a/src/docforge/changesets.py b/src/docforge/changesets.py index 74412ce..d4c878e 100644 --- a/src/docforge/changesets.py +++ b/src/docforge/changesets.py @@ -1215,6 +1215,11 @@ class ChangesetStore: def _restore(path: Path, previous: bytes | None, root: Path) -> None: if previous is None: path.unlink(missing_ok=True) + directory_descriptor = os.open(root, os.O_RDONLY) + try: + os.fsync(directory_descriptor) + finally: + os.close(directory_descriptor) return restore_descriptor, restore_name = tempfile.mkstemp(prefix=".rollback-", dir=root) restore = Path(restore_name) @@ -1224,6 +1229,11 @@ class ChangesetStore: handle.flush() os.fsync(handle.fileno()) os.replace(restore, path) + directory_descriptor = os.open(root, os.O_RDONLY) + try: + os.fsync(directory_descriptor) + finally: + os.close(directory_descriptor) except Exception: restore.unlink(missing_ok=True) raise diff --git a/src/docforge/incremental.py b/src/docforge/incremental.py index 7f5f511..9853965 100644 --- a/src/docforge/incremental.py +++ b/src/docforge/incremental.py @@ -5,11 +5,11 @@ from __future__ import annotations import json import os import stat -import tempfile from dataclasses import dataclass from pathlib import Path from typing import Any, cast +from ._fs_safety import atomic_replace_bytes_at, open_bound_directory, require_bound_directory from .errors import DocForgeError EXTRACTION_CACHE_SCHEMA_VERSION = 1 @@ -186,24 +186,23 @@ def write_extraction_cache( maximum=max_bytes, actual=len(encoded), ) - with tempfile.NamedTemporaryFile( - mode="wb", - prefix="extractions-", - suffix=".json", - dir=path.parent, - delete=False, - ) as descriptor: - temporary = Path(descriptor.name) - descriptor.write(encoded) - descriptor.flush() - os.fsync(descriptor.fileno()) + directory_fd = open_bound_directory(path.parent) try: - os.replace(temporary, path) - except OSError as error: - temporary.unlink(missing_ok=True) + atomic_replace_bytes_at( + path.parent, + directory_fd, + path.name, + encoded, + verify=lambda: require_bound_directory(path.parent, directory_fd), + ) + except DocForgeError as error: + if error.code == "path_escape": + raise raise DocForgeError( "cache_failure", "Could not publish the incremental extraction cache" ) from error + finally: + os.close(directory_fd) def affected_sources( diff --git a/src/docforge/rendering.py b/src/docforge/rendering.py index 86738d3..e224c3c 100644 --- a/src/docforge/rendering.py +++ b/src/docforge/rendering.py @@ -13,6 +13,12 @@ from contextlib import contextmanager from pathlib import Path from typing import cast +from ._fs_safety import ( + atomic_replace_bytes_at, + open_bound_directory, + open_confined_directory, + require_bound_directory, +) from .changesets import ChangesetStore from .errors import DocForgeError from .models import ( @@ -881,22 +887,30 @@ class RenderService: parent.mkdir(parents=True, exist_ok=True) if parent.resolve() != parent or not parent.is_relative_to(root): raise DocForgeError("path_escape", "Render output directory is unsafe") - descriptor, temporary_name = tempfile.mkstemp(prefix=".docforge-render-", dir=parent) - temporary = Path(temporary_name) - try: - with os.fdopen(descriptor, "wb") as handle: - handle.write(content) - handle.flush() - os.fsync(handle.fileno()) + directory_fd = ( + open_bound_directory(root) + if parent == root + else open_confined_directory(root, parent, create=False) + ) + + def verify_bound() -> None: verify() - if output.is_symlink(): - raise DocForgeError("path_escape", "Render output became unsafe") - os.replace(temporary, output) + require_bound_directory(parent, directory_fd) + + try: + atomic_replace_bytes_at( + parent, + directory_fd, + output.name, + content, + verify=verify_bound, + ) except Exception: - temporary.unlink(missing_ok=True) if preview: self._remove_empty_preview_parents(parent) raise + finally: + os.close(directory_fd) def _remove_empty_preview_parents(self, parent: Path) -> None: config = self.project.descriptor.render diff --git a/tests/test_changesets.py b/tests/test_changesets.py index c294abf..bfd878f 100644 --- a/tests/test_changesets.py +++ b/tests/test_changesets.py @@ -3,13 +3,16 @@ from __future__ import annotations import hashlib import json import multiprocessing +import os import shutil +import stat import tempfile import unittest from pathlib import Path from typing import Any from unittest import mock +import docforge.application as application_module from docforge.application import CanonicalApplicationService, GenericCanonicalApplier from docforge.changesets import ChangesetStore from docforge.errors import DocForgeError @@ -271,6 +274,291 @@ class DocForgeChangesetTests(unittest.TestCase): service.apply("apply-all", str(final["changeset_hash"])) self.assertEqual("changeset_closed", closed.exception.code) + def test_canonical_update_exchange_preserves_a_raced_external_edit(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = self.copy_fixture(Path(directory)) + project = Project.open(root) + store = ChangesetStore(project, "alpha-editor") + proposal = store.register( + "update-race", + [ + { + "operation": "update", + "node_id": "guide.workflow", + "metadata": {"summary": "Approved summary."}, + "rationale": "Exercise the atomic update boundary.", + } + ], + ) + proposal_path = root / ".docforge/changesets/update-race.json" + proposal_bytes = proposal_path.read_bytes() + target = root / "docs/content/workflow.md" + exchange = application_module.rename_exchange_at + raced = False + + def race(directory_fd: int, first: str, second: str) -> None: + nonlocal raced + if second == target.name and not raced: + raced = True + target.write_bytes(target.read_bytes() + b"\nExternal edit at exchange.\n") + exchange(directory_fd, first, second) + + with ( + mock.patch( + "docforge.application.rename_exchange_at", + side_effect=race, + ), + self.assertRaises(DocForgeError) as captured, + ): + store.apply( + changeset_id="update-race", + expected_changeset_hash=str(proposal["changeset_hash"]), + applier_id="alpha-editor", + application=GenericCanonicalApplier(project).apply, + ) + + self.assertEqual("base_conflict", captured.exception.code) + self.assertIn("External edit at exchange.", target.read_text(encoding="utf-8")) + self.assertEqual(proposal_bytes, proposal_path.read_bytes()) + self.assertFalse((root / ".docforge/changesets/.state/update-race.json").exists()) + self.assertFalse(tuple(target.parent.glob(".docforge-apply-*"))) + + def test_canonical_create_and_delete_races_preserve_foreign_targets(self) -> None: + with tempfile.TemporaryDirectory() as directory: + parent = Path(directory) + + create_root = self.copy_fixture(parent / "create") + create_project = Project.open(create_root) + create_store = ChangesetStore(create_project, "alpha-editor") + create = create_store.register( + "create-race", + [ + { + "operation": "create", + "node_id": "guide.raced", + "target_source": "docs/content/raced.md", + "metadata": self.new_metadata(), + "content": "Approved new content.", + "rationale": "Exercise no-replace creation.", + } + ], + ) + create_target = create_root / "docs/content/raced.md" + real_link = application_module.os.link + appeared = False + + def race_create( + source: str, + target: str, + *, + src_dir_fd: int, + dst_dir_fd: int, + follow_symlinks: bool, + ) -> None: + nonlocal appeared + if target == create_target.name and not appeared: + appeared = True + create_target.write_bytes(b"foreign create target\n") + real_link( + source, + target, + src_dir_fd=src_dir_fd, + dst_dir_fd=dst_dir_fd, + follow_symlinks=follow_symlinks, + ) + + with ( + mock.patch("docforge.application.os.link", side_effect=race_create), + self.assertRaises(DocForgeError) as create_error, + ): + create_store.apply( + changeset_id="create-race", + expected_changeset_hash=str(create["changeset_hash"]), + applier_id="alpha-editor", + application=GenericCanonicalApplier(create_project).apply, + ) + self.assertEqual("base_conflict", create_error.exception.code) + self.assertEqual(b"foreign create target\n", create_target.read_bytes()) + self.assertFalse(tuple(create_target.parent.glob(".docforge-apply-*"))) + + delete_root = self.copy_fixture(parent / "delete") + delete_project = Project.open(delete_root) + delete_store = ChangesetStore(delete_project, "alpha-editor") + delete = delete_store.register( + "delete-race", + [ + { + "operation": "delete", + "node_id": "proof.validation", + "relationship_changes": [ + { + "action": "remove", + "source_id": "proof.validation", + "relation": "proves", + "target_id": "guide.workflow", + } + ], + "rationale": "Exercise atomic deletion.", + } + ], + ) + delete_target = delete_root / "docs/content/proof.toml" + exchange = application_module.rename_exchange_at + deleted_race = False + + def race_delete(directory_fd: int, first: str, second: str) -> None: + nonlocal deleted_race + if second == delete_target.name and not deleted_race: + deleted_race = True + delete_target.write_bytes( + delete_target.read_bytes() + b"\n# foreign delete edit\n" + ) + exchange(directory_fd, first, second) + + with ( + mock.patch( + "docforge.application.rename_exchange_at", + side_effect=race_delete, + ), + self.assertRaises(DocForgeError) as delete_error, + ): + delete_store.apply( + changeset_id="delete-race", + expected_changeset_hash=str(delete["changeset_hash"]), + applier_id="alpha-editor", + application=GenericCanonicalApplier(delete_project).apply, + ) + self.assertEqual("base_conflict", delete_error.exception.code) + self.assertIn("# foreign delete edit", delete_target.read_text(encoding="utf-8")) + self.assertFalse(tuple(delete_target.parent.glob(".docforge-apply-*"))) + + def test_rollback_never_clobbers_a_foreign_edit_and_retains_original_bytes(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = self.copy_fixture(Path(directory)) + project = Project.open(root) + store = ChangesetStore(project, "alpha-editor") + proposal = store.register( + "rollback-race", + [ + { + "operation": "update", + "node_id": "guide.foundation", + "metadata": {"summary": "First approved update."}, + "rationale": "Publish before the synthetic failure.", + }, + { + "operation": "update", + "node_id": "guide.workflow", + "metadata": {"summary": "Second approved update."}, + "rationale": "Trigger rollback after the first publication.", + }, + ], + ) + first_target = root / "docs/content/foundation.md" + first_before = first_target.read_bytes() + publish = GenericCanonicalApplier._publish + calls = 0 + + def fail_after_foreign_edit( + applier: GenericCanonicalApplier, + publication: Any, + ) -> None: + nonlocal calls + calls += 1 + if calls == 1: + publish(applier, publication) + first_target.write_bytes( + first_target.read_bytes() + b"\nForeign edit after publication.\n" + ) + return + raise DocForgeError("application_failure", "Synthetic second-target failure") + + with ( + mock.patch.object( + GenericCanonicalApplier, + "_publish", + autospec=True, + side_effect=fail_after_foreign_edit, + ), + self.assertRaises(DocForgeError) as captured, + ): + store.apply( + changeset_id="rollback-race", + expected_changeset_hash=str(proposal["changeset_hash"]), + applier_id="alpha-editor", + application=GenericCanonicalApplier(project).apply, + ) + + self.assertEqual("application_recovery_required", captured.exception.code) + conflicts = captured.exception.details["conflicts"] + self.assertEqual("target_or_backup_changed", conflicts[0]["reason"]) + self.assertIn( + "Foreign edit after publication.", + first_target.read_text(encoding="utf-8"), + ) + retained = root / conflicts[0]["retained"] + self.assertTrue(retained.is_file()) + self.assertEqual(first_before, retained.read_bytes()) + self.assertFalse((root / ".docforge/changesets/.state/rollback-race.json").exists()) + + def test_canonical_update_preserves_existing_file_mode(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = self.copy_fixture(Path(directory)) + target = root / "docs/content/workflow.md" + target.chmod(0o640) + project = Project.open(root) + store = ChangesetStore(project, "alpha-editor") + proposal = store.register( + "mode", + [ + { + "operation": "update", + "node_id": "guide.workflow", + "metadata": {"summary": "Mode-preserving update."}, + "rationale": "Preserve canonical file permissions.", + } + ], + ) + + result = store.apply( + changeset_id="mode", + expected_changeset_hash=str(proposal["changeset_hash"]), + applier_id="alpha-editor", + application=GenericCanonicalApplier(project).apply, + ) + + self.assertTrue(result["applied"]) + self.assertEqual(0o640, stat.S_IMODE(target.stat().st_mode)) + self.assertEqual([], result["retained_recovery_files"]) + + def test_changeset_rollback_fsyncs_the_parent_directory(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + target = root / "proposal.json" + real_fsync = os.fsync + fsynced_modes: list[int] = [] + + def record_fsync(descriptor: int) -> None: + fsynced_modes.append(os.fstat(descriptor).st_mode) + real_fsync(descriptor) + + for previous in (b"previous proposal\n", None): + with self.subTest(previous=previous): + target.write_bytes(b"replacement proposal\n") + fsynced_modes.clear() + + with mock.patch( + "docforge.changesets.os.fsync", + side_effect=record_fsync, + ): + ChangesetStore._restore(target, previous, root) + + self.assertTrue(any(stat.S_ISDIR(mode) for mode in fsynced_modes)) + if previous is None: + self.assertFalse(target.exists()) + else: + self.assertEqual(previous, target.read_bytes()) + def test_abandoned_proposal_releases_overlap_and_stale_work_remains_active(self) -> None: with tempfile.TemporaryDirectory() as directory: root = self.copy_fixture(Path(directory)) diff --git a/tests/test_incremental_cache.py b/tests/test_incremental_cache.py index cfd2a9d..773f1c6 100644 --- a/tests/test_incremental_cache.py +++ b/tests/test_incremental_cache.py @@ -2,9 +2,11 @@ from __future__ import annotations import hashlib import os +import stat import tempfile import unittest from pathlib import Path +from unittest import mock from docforge.errors import DocForgeError from docforge.incremental import ( @@ -97,6 +99,24 @@ class ExtractionCacheBoundsTests(unittest.TestCase): ) ) + def test_cache_publication_fsyncs_the_parent_directory(self) -> None: + with tempfile.TemporaryDirectory() as directory: + path = Path(directory) / "extractions.json" + real_fsync = os.fsync + fsynced_modes: list[int] = [] + + def record_fsync(descriptor: int) -> None: + fsynced_modes.append(os.fstat(descriptor).st_mode) + real_fsync(descriptor) + + with mock.patch( + "docforge._fs_safety.os.fsync", + side_effect=record_fsync, + ): + write_extraction_cache(path, self.cache(), max_bytes=1_000, max_sources=2) + + self.assertTrue(any(stat.S_ISDIR(mode) for mode in fsynced_modes)) + if __name__ == "__main__": unittest.main() diff --git a/tests/test_rendering.py b/tests/test_rendering.py index e37b837..843de96 100644 --- a/tests/test_rendering.py +++ b/tests/test_rendering.py @@ -4,6 +4,7 @@ import contextlib import hashlib import io import json +import os import shutil import tempfile import unittest @@ -302,6 +303,49 @@ class DocForgeRenderingTests(unittest.TestCase): self.assertEqual(committed_before, committed_output.read_bytes()) self.assertEqual("current", service.status("manual")["state"]) + def test_manual_and_preview_publication_fsync_their_directories(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = self.copy_fixture("alpha", Path(directory)) + project = Project.open(root) + changesets = ChangesetStore(project, "alpha-editor") + service = RenderService(project, changesets) + proposal = changesets.create("durable-preview") + proposal = changesets.propose_update( + changeset_id="durable-preview", + expected_changeset_hash=str(proposal["changeset_hash"]), + node_id="guide.workflow", + expected_content_hash=self.node_hash(project, "guide.workflow"), + metadata={"summary": "Durable preview output."}, + content=None, + relationship_changes=[], + rationale="Exercise durable preview publication.", + ) + del proposal + real_fsync = os.fsync + fsynced_directories: set[Path] = set() + + def record_fsync(descriptor: int) -> None: + try: + path = Path(os.readlink(f"/proc/self/fd/{descriptor}")) + if path.is_dir(): + fsynced_directories.add(path) + except OSError: + pass + real_fsync(descriptor) + + with mock.patch( + "docforge._fs_safety.os.fsync", + side_effect=record_fsync, + ): + service.render("manual") + service.preview("durable-preview", "manual") + + self.assertIn(root / ".docforge/rendered", fsynced_directories) + self.assertIn( + root / ".docforge/previews/durable-preview", + fsynced_directories, + ) + def test_failed_and_mid_input_renders_preserve_previous_outputs(self) -> None: with tempfile.TemporaryDirectory() as directory: root = self.copy_fixture("alpha", Path(directory))