"""Confined preview and derived-output orchestration for declared render views.""" from __future__ import annotations import fcntl import hashlib import json import os import stat import tempfile from collections.abc import Callable, Generator from contextlib import contextmanager from pathlib import Path from typing import cast from .changesets import ChangesetStore from .errors import DocForgeError from .models import ( GenerationRecordingProject, IncrementalStateProject, ProjectDescriptor, ProjectService, ProjectSnapshot, ProjectState, RenderConfig, RenderView, ) from .project import project_root_fingerprint from .projection_contract import ProjectionReceiptV1 from .projection_policy import ManualProjectionMode, validate_manual_projection_mode from .render_contract import PreparedRender, relative_output, renderer_for from .telemetry import increment, stage RENDER_RECEIPT_SCHEMA_VERSION = 1 MAX_RENDER_RECEIPT_BYTES = 64_000 class RenderService: """Render only declared views through fixed built-in renderer implementations.""" def __init__( self, project: ProjectService, changesets: ChangesetStore | None = None, *, manual_policy: ManualProjectionMode = "explicit", ) -> None: self.project = project self.changesets = changesets or ChangesetStore(project) self.manual_policy = validate_manual_projection_mode(manual_policy) def _require_rendering(self, operation: str) -> None: if self.manual_policy == "disabled": raise DocForgeError( "projection_policy_forbids_operation", "Manual projection policy disables rendering work", projection="manual", mode=self.manual_policy, operation=operation, ) def status(self, view_id: str | None = None) -> dict[str, object]: """Report publication state from bounded receipts without rendering canonical content.""" with stage("render.status"): return self._status(view_id) def _status(self, view_id: str | None = None) -> dict[str, object]: descriptor = self.project.descriptor config = descriptor.render current_state = self._current_state() if config is None: return self._status_result( descriptor, current_state, configured=False, state="not_configured", verification="receipt", outputs=[], ) views = self._views(config, view_id) first_outputs = [self._receipt_status(descriptor, view, current_state) for view in views] outputs = [self._receipt_status(descriptor, view, current_state) for view in views] if outputs != first_outputs: for output in outputs: if output["state"] == "current": output["state"] = "stale" output["reason"] = "publication_changed_during_status" final_state = self._current_state() if final_state != current_state: for output in outputs: if output["state"] == "current": output["state"] = "stale" output["reason"] = "source_changed_during_status" identity = final_state if final_state is not None else current_state return self._status_result( descriptor, identity, configured=True, state="current" if all(item["state"] == "current" for item in outputs) else "stale", verification="receipt", outputs=outputs, ) def deep_status(self, view_id: str | None = None) -> dict[str, object]: """Recompute render output as the explicit side-effect-free equivalence oracle.""" self._require_rendering("deep_status") snapshot = self.project.load() config = snapshot.descriptor.render if config is None: return self._result( snapshot, configured=False, state="not_configured", verification="deep", outputs=[], ) views = self._views(config, view_id) outputs: list[dict[str, object]] = [] for view in views: template_before = self._safe_file_identity( snapshot.descriptor.root, view.template_path, ) output_before = self._safe_file_identity( snapshot.descriptor.root, view.output_path, ) prepared, _ = self._prepare( snapshot, view, changeset_hash=None, incremental=False, ) state = "missing" actual_hash: str | None = None output = view.output_path if output.is_symlink() or output.resolve(strict=False) != output: state = "unsafe" elif output.is_file(): if output.stat().st_size > snapshot.descriptor.limits.max_render_bytes: state = "oversized" else: raw = output.read_bytes() increment("render_output_bytes_hashed", len(raw)) with stage("render.output_hash"): actual_hash = hashlib.sha256(raw).hexdigest() state = "current" if actual_hash == prepared.output_hash else "stale" result = self._view_result( snapshot, view, prepared, state=state, actual_hash=actual_hash, ) if template_before != self._safe_file_identity( snapshot.descriptor.root, view.template_path ) or output_before != self._safe_file_identity( snapshot.descriptor.root, view.output_path ): result["state"] = "stale" result["reason"] = "publication_changed_during_deep_status" outputs.append(result) current = self.project.load() if current.source_hash != snapshot.source_hash or current.revision != snapshot.revision: for output in outputs: output["state"] = "stale" output["reason"] = "source_changed_during_deep_status" return self._result( snapshot, configured=True, state="current" if all(item["state"] == "current" for item in outputs) else "stale", verification="deep", outputs=outputs, ) def render(self, view_id: str) -> dict[str, object]: self._require_rendering("render") with self._lock(): snapshot = self.project.load() config = self._config(snapshot) view = self._views(config, view_id)[0] prepared, template_bytes = self._prepare(snapshot, view, changeset_hash=None) self._atomic_write( view.output_path, prepared.output, verify=lambda: self._verify_canonical(snapshot, view, template_bytes), ) receipt: dict[str, object] state = "current" try: if isinstance(self.project, GenerationRecordingProject): self.project.record_generation(snapshot) receipt = self._publish_receipt(snapshot, view, prepared) except (DocForgeError, OSError) as error: state = "degraded" receipt = { "state": "failed", "error": ( error.as_dict() if isinstance(error, DocForgeError) else { "code": "render_receipt_failure", "message": "Rendered output was published but its receipt failed", "details": {}, } ), } return self._result( snapshot, configured=True, state=state, publication="published", receipt=receipt, output=self._view_result( snapshot, view, prepared, state="current", actual_hash=prepared.output_hash, ), ) def _receipt_status( self, descriptor: ProjectDescriptor, view: RenderView, current_state: ProjectState | None, ) -> dict[str, object]: receipt, receipt_state = self._read_receipt(view) output_state = self._safe_file_identity(descriptor.root, view.output_path) if output_state is None: state = "unsafe" if view.output_path.is_symlink() else "missing" return self._receipt_view_result( descriptor, view, receipt, state=state, reason="output_not_safe" if state == "unsafe" else "output_missing", ) if receipt is None: return self._receipt_view_result( descriptor, view, receipt, state="unverified", reason=receipt_state, ) if not self._receipt_matches_binding(descriptor, view, receipt): return self._receipt_view_result( descriptor, view, receipt, state="unverified", reason="foreign_or_incompatible_receipt", ) template_state = self._safe_file_identity(descriptor.root, view.template_path) if template_state is None: return self._receipt_view_result( descriptor, view, receipt, state="unsafe", reason="template_not_safe", ) if receipt.get("template_file") != template_state: return self._receipt_view_result( descriptor, view, receipt, state="stale", reason="template_changed", ) if receipt.get("output_file") != output_state: return self._receipt_view_result( descriptor, view, receipt, state="stale", reason="output_changed", ) if current_state is None: reason = ( "source_generation_unavailable" if isinstance(self.project, GenerationRecordingProject) else "source_generation_unsupported" ) return self._receipt_view_result( descriptor, view, receipt, state=( "stale" if isinstance(self.project, GenerationRecordingProject) else "unverified" ), reason=reason, ) if ( receipt.get("source_hash") != current_state.source_hash or receipt.get("revision") != current_state.revision ): return self._receipt_view_result( descriptor, view, receipt, state="stale", reason="source_generation_changed", ) return self._receipt_view_result( descriptor, view, receipt, state="current", reason=None, ) def _publish_receipt( self, snapshot: ProjectSnapshot, view: RenderView, prepared: PreparedRender, ) -> dict[str, object]: source_before = self._current_state() if isinstance(self.project, GenerationRecordingProject) and ( source_before is None or source_before.source_hash != snapshot.source_hash or source_before.revision != snapshot.revision ): raise DocForgeError( "render_receipt_failure", "Canonical source changed before render receipt publication", ) if source_before is not None and ( source_before.source_hash != snapshot.source_hash or source_before.revision != snapshot.revision ): raise DocForgeError( "render_receipt_failure", "Canonical source changed before render receipt publication", ) template_file, template_hash = self._verified_file_digest( snapshot.descriptor.root, view.template_path, snapshot.descriptor.limits.max_template_bytes, ) output_file, output_hash = self._verified_file_digest( snapshot.descriptor.root, view.output_path, snapshot.descriptor.limits.max_render_bytes, ) if ( template_hash != prepared.template_hash or output_hash != prepared.output_hash or output_file["size"] != len(prepared.output) ): raise DocForgeError( "render_receipt_failure", "Published render files do not match the verified render", ) final_template_file, final_template_hash = self._verified_file_digest( snapshot.descriptor.root, view.template_path, snapshot.descriptor.limits.max_template_bytes, ) final_output_file, final_output_hash = self._verified_file_digest( snapshot.descriptor.root, view.output_path, snapshot.descriptor.limits.max_render_bytes, ) source_after = self._current_state() if ( template_file != final_template_file or output_file != final_output_file or template_hash != final_template_hash or output_hash != final_output_hash or source_before != source_after ): raise DocForgeError( "render_receipt_failure", "Render publication changed while its receipt was being prepared", ) payload: dict[str, object] = { "schema_version": RENDER_RECEIPT_SCHEMA_VERSION, "project_id": snapshot.descriptor.project_id, "project_root_fingerprint": project_root_fingerprint(snapshot.descriptor.root), "adapter": snapshot.descriptor.adapter, "revision": snapshot.revision, "source_hash": snapshot.source_hash, "view_id": view.view_id, "view_config_hash": self._view_config_hash(snapshot.descriptor, view), "renderer": prepared.renderer, "renderer_version": prepared.renderer_version, "render_identity": prepared.render_identity, "template_hash": prepared.template_hash, "output_hash": prepared.output_hash, "output_bytes": len(prepared.output), "template_file": final_template_file, "output_file": final_output_file, } if prepared.projection_receipt is not None: payload["projection_receipt"] = prepared.projection_receipt raw = json.dumps(payload, sort_keys=True, indent=2).encode("utf-8") + b"\n" if len(raw) > MAX_RENDER_RECEIPT_BYTES: raise DocForgeError( "render_receipt_failure", "Render publication receipt exceeds its fixed size limit", ) root = self._receipt_root(create=True) path = root / f"{view.view_id}.json" if path.is_symlink(): raise DocForgeError( "path_escape", "Render publication receipt path is not safe", ) descriptor, temporary_name = tempfile.mkstemp(prefix=".render-receipt-", dir=root) temporary = Path(temporary_name) try: with os.fdopen(descriptor, "wb") as handle: handle.write(raw) handle.flush() os.fsync(handle.fileno()) last_template_file, last_template_hash = self._verified_file_digest( snapshot.descriptor.root, view.template_path, snapshot.descriptor.limits.max_template_bytes, ) last_output_file, last_output_hash = self._verified_file_digest( snapshot.descriptor.root, view.output_path, snapshot.descriptor.limits.max_render_bytes, ) if ( last_template_file != final_template_file or last_output_file != final_output_file or last_template_hash != final_template_hash or last_output_hash != final_output_hash or self._current_state() != source_after ): raise DocForgeError( "render_receipt_failure", "Render publication changed before receipt publication", ) os.replace(temporary, path) directory_descriptor = os.open(root, os.O_RDONLY) try: os.fsync(directory_descriptor) finally: os.close(directory_descriptor) except Exception: temporary.unlink(missing_ok=True) raise return { "state": "current", "schema_version": RENDER_RECEIPT_SCHEMA_VERSION, "path": path.relative_to(snapshot.descriptor.root).as_posix(), } def _read_receipt( self, view: RenderView, ) -> tuple[dict[str, object] | None, str]: try: root = self._receipt_root(create=False) except DocForgeError: return None, "receipt_root_unsafe" path = root / f"{view.view_id}.json" if path.is_symlink(): return None, "receipt_unsafe" if not path.is_file(): return None, "receipt_missing" try: if path.stat().st_size > MAX_RENDER_RECEIPT_BYTES: return None, "receipt_oversized" raw = path.read_bytes() if len(raw) > MAX_RENDER_RECEIPT_BYTES: return None, "receipt_oversized" parsed: object = json.loads(raw) except (OSError, UnicodeDecodeError, json.JSONDecodeError): return None, "receipt_corrupt" if not isinstance(parsed, dict): return None, "receipt_corrupt" return cast(dict[str, object], parsed), "receipt" def _receipt_root(self, *, create: bool) -> Path: cache_root = self.project.descriptor.cache_root root = cache_root / "render-receipts" if ( cache_root.resolve(strict=False) != cache_root or root.is_symlink() or root.resolve(strict=False) != root or not root.is_relative_to(cache_root) ): raise DocForgeError("path_escape", "Render receipt root is not safe") if create: cache_root.mkdir(parents=True, exist_ok=True) root.mkdir(parents=True, exist_ok=True) if root.exists() and not root.is_dir(): raise DocForgeError("path_escape", "Render receipt root is not safe") return root @staticmethod def _safe_file_identity(root: Path, path: Path) -> dict[str, object] | None: if path.is_symlink() or path.resolve(strict=False) != path or not path.is_relative_to(root): return None try: current = path.lstat() except OSError: return None if not stat.S_ISREG(current.st_mode): return None return { "path": path.relative_to(root).as_posix(), "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, } @staticmethod def _verified_file_digest( root: Path, path: Path, maximum: int, ) -> tuple[dict[str, object], str]: if path.is_symlink() or path.resolve(strict=False) != path or not path.is_relative_to(root): raise DocForgeError( "render_receipt_failure", "Render publication file is not safe for verification", ) try: descriptor = os.open(path, os.O_RDONLY | os.O_NOFOLLOW) except OSError as error: raise DocForgeError( "render_receipt_failure", "Render publication file is not readable for verification", ) 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: raise DocForgeError( "render_receipt_failure", "Render publication file failed receipt validation", ) digest = hashlib.file_digest(handle, "sha256").hexdigest() return ( { "path": path.relative_to(root).as_posix(), "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, }, digest, ) @staticmethod def _view_config_hash(descriptor: ProjectDescriptor, view: RenderView) -> str: payload = { "view_id": view.view_id, "renderer": view.renderer, "template": view.template_path.relative_to(descriptor.root).as_posix(), "output": view.output_path.relative_to(descriptor.root).as_posix(), "title": view.title, "families": list(view.families), } return hashlib.sha256( json.dumps(payload, sort_keys=True, separators=(",", ":")).encode("utf-8") ).hexdigest() def _receipt_matches_binding( self, descriptor: ProjectDescriptor, view: RenderView, receipt: dict[str, object], ) -> bool: required = { "schema_version", "project_id", "project_root_fingerprint", "adapter", "revision", "source_hash", "view_id", "view_config_hash", "renderer", "renderer_version", "render_identity", "template_hash", "output_hash", "output_bytes", "template_file", "output_file", } renderer = renderer_for(view) template_file = receipt.get("template_file") output_file = receipt.get("output_file") fields = set(receipt) projection_receipt = receipt.get("projection_receipt") return ( fields in (required, required | {"projection_receipt"}) and receipt.get("schema_version") == RENDER_RECEIPT_SCHEMA_VERSION and receipt.get("project_id") == descriptor.project_id and receipt.get("project_root_fingerprint") == project_root_fingerprint(descriptor.root) and receipt.get("adapter") == descriptor.adapter and receipt.get("view_id") == view.view_id and receipt.get("view_config_hash") == self._view_config_hash(descriptor, view) and receipt.get("renderer") == renderer.renderer_id and receipt.get("renderer_version") == renderer.renderer_version and self._is_hash(receipt.get("source_hash")) and isinstance(receipt.get("revision"), str) and bool(receipt.get("revision")) and self._is_hash(receipt.get("view_config_hash")) and self._is_hash(receipt.get("render_identity")) and self._is_hash(receipt.get("template_hash")) and self._is_hash(receipt.get("output_hash")) and type(receipt.get("output_bytes")) is int and 0 <= cast(int, receipt["output_bytes"]) <= descriptor.limits.max_render_bytes and self._valid_receipt_file( template_file, view.template_path.relative_to(descriptor.root).as_posix(), descriptor.limits.max_template_bytes, ) and self._valid_receipt_file( output_file, view.output_path.relative_to(descriptor.root).as_posix(), descriptor.limits.max_render_bytes, ) and cast(dict[str, object], output_file)["size"] == receipt.get("output_bytes") and ( projection_receipt is None or self._valid_projection_receipt( projection_receipt, renderer_id=renderer.renderer_id, renderer_version=renderer.renderer_version, output_hash=cast(str, receipt["output_hash"]), output_bytes=cast(int, receipt["output_bytes"]), ) ) ) @staticmethod def _valid_projection_receipt( value: object, *, renderer_id: str, renderer_version: str, output_hash: str, output_bytes: int, ) -> bool: if not isinstance(value, dict): return False try: receipt = ProjectionReceiptV1.from_dict(cast(dict[str, object], value)) except DocForgeError: return False document = receipt.document return ( document.get("kind") == "manual" and document.get("renderer") == { "renderer_id": renderer_id, "renderer_version": renderer_version, } and document.get("artifacts") == [ { "artifact_id": "manual.html", "media_type": "text/html; charset=utf-8", "sha256": output_hash, "bytes": output_bytes, } ] ) @staticmethod def _is_hash(value: object) -> bool: return ( isinstance(value, str) and len(value) == 64 and all(character in "0123456789abcdef" for character in value) ) @staticmethod def _valid_receipt_file( value: object, expected_path: str, maximum: int, ) -> bool: if not isinstance(value, dict): return False payload = cast(dict[str, object], value) return ( set(payload) == { "path", "device", "inode", "mode", "size", "mtime_ns", "ctime_ns", } and payload.get("path") == expected_path and all( type(payload.get(key)) is int and cast(int, payload[key]) >= 0 for key in ("device", "inode", "mode", "size", "mtime_ns", "ctime_ns") ) and cast(int, payload["size"]) <= maximum ) @staticmethod def _receipt_view_result( descriptor: ProjectDescriptor, view: RenderView, receipt: dict[str, object] | None, *, state: str, reason: str | None, ) -> dict[str, object]: payload = receipt or {} return { "view_id": view.view_id, "renderer": payload.get("renderer", view.renderer), "renderer_version": payload.get("renderer_version"), "render_identity": payload.get("render_identity"), "expected_output_hash": payload.get("output_hash"), "actual_output_hash": (payload.get("output_hash") if state == "current" else None), "template_hash": payload.get("template_hash"), "path": view.output_path.relative_to(descriptor.root).as_posix(), "state": state, "reason": reason, "verification": "receipt", "receipt_schema_version": payload.get("schema_version"), "projection_receipt": payload.get("projection_receipt"), } def _current_state(self) -> ProjectState | None: if isinstance(self.project, IncrementalStateProject): return self.project.incremental_state() return None @staticmethod def _status_result( descriptor: ProjectDescriptor, identity: ProjectState | None, **payload: object, ) -> dict[str, object]: return { "status": "ok", "project_id": descriptor.project_id, "project_root_fingerprint": project_root_fingerprint(descriptor.root), "adapter": descriptor.adapter, "revision": identity.revision if identity is not None else "unknown", "source_hash": identity.source_hash if identity is not None else None, **payload, } def preview(self, changeset_id: str, view_id: str) -> dict[str, object]: self._require_rendering("preview") with self._lock(): snapshot, changeset_hash = self.changesets.projected_snapshot(changeset_id) config = self._config(snapshot) view = self._views(config, view_id)[0] prepared, template_bytes = self._prepare(snapshot, view, changeset_hash=changeset_hash) preview_path = config.preview_root / changeset_id / f"{view.view_id}.html" if not preview_path.is_relative_to(config.preview_root): raise DocForgeError("path_escape", "Preview path escaped its configured root") def verify() -> None: current, current_hash = self.changesets.projected_snapshot(changeset_id) if current.source_hash != snapshot.source_hash or current_hash != changeset_hash: raise DocForgeError( "render_input_changed", "Changeset or canonical input changed during preview", ) self._verify_template(view, template_bytes) self._atomic_write(preview_path, prepared.output, verify=verify, preview=True) return self._result( snapshot, configured=True, state="current", changeset_id=changeset_id, changeset_hash=changeset_hash, preview_identity=prepared.render_identity, preview={ **self._view_result( snapshot, view, prepared, state="current", actual_hash=prepared.output_hash, ), "path": relative_output(snapshot, preview_path), }, ) def _prepare( self, snapshot: ProjectSnapshot, view: RenderView, *, changeset_hash: str | None, incremental: bool = True, ) -> tuple[PreparedRender, bytes]: increment("render_prepare_calls") template = self._template_bytes(snapshot, view) with stage("render.prepare"): prepared = renderer_for(view, incremental=incremental).prepare( snapshot, view, template, changeset_hash=changeset_hash, ) increment("render_output_bytes_built", len(prepared.output)) if len(prepared.output) > snapshot.descriptor.limits.max_render_bytes: raise DocForgeError("render_too_large", "Rendered output exceeds the configured limit") return prepared, template def _template_bytes(self, snapshot: ProjectSnapshot, view: RenderView) -> bytes: path = view.template_path config = self._config(snapshot) if ( path.is_symlink() or path.resolve(strict=False) != path or config.template_root.resolve(strict=False) != config.template_root or not path.is_file() or not path.is_relative_to(config.template_root) ): raise DocForgeError("unsafe_template", "Render template is missing or unsafe") if path.stat().st_size > snapshot.descriptor.limits.max_template_bytes: raise DocForgeError( "template_too_large", "Render template exceeds the configured limit" ) raw = path.read_bytes() if len(raw) > snapshot.descriptor.limits.max_template_bytes: raise DocForgeError( "template_too_large", "Render template exceeds the configured limit" ) return raw def _verify_canonical( self, snapshot: ProjectSnapshot, view: RenderView, template_bytes: bytes ) -> None: current = self.project.load() if current.source_hash != snapshot.source_hash: raise DocForgeError("render_input_changed", "Canonical input changed during rendering") self._verify_template(view, template_bytes) @staticmethod def _verify_template(view: RenderView, template_bytes: bytes) -> None: if ( view.template_path.is_symlink() or view.template_path.resolve(strict=False) != view.template_path or not view.template_path.is_file() ): raise DocForgeError("render_input_changed", "Render template changed during rendering") if view.template_path.read_bytes() != template_bytes: raise DocForgeError("render_input_changed", "Render template changed during rendering") def _atomic_write( self, output: Path, content: bytes, *, verify: Callable[[], None], preview: bool = False, ) -> None: root = self.project.descriptor.root if output.is_symlink() or not output.is_relative_to(root): raise DocForgeError("path_escape", "Render output path is unsafe") parent = output.parent if parent.resolve(strict=False) != parent: raise DocForgeError("path_escape", "Render output directory is unsafe") 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()) verify() if output.is_symlink(): raise DocForgeError("path_escape", "Render output became unsafe") os.replace(temporary, output) except Exception: temporary.unlink(missing_ok=True) if preview: self._remove_empty_preview_parents(parent) raise def _remove_empty_preview_parents(self, parent: Path) -> None: config = self.project.descriptor.render if config is None: return current = parent while current != config.preview_root: try: current.rmdir() except OSError: return current = current.parent def _view_result( self, snapshot: ProjectSnapshot, view: RenderView, prepared: PreparedRender, *, state: str, actual_hash: str | None, ) -> dict[str, object]: return { "view_id": view.view_id, "renderer": prepared.renderer, "renderer_version": prepared.renderer_version, "render_identity": prepared.render_identity, "expected_output_hash": prepared.output_hash, "actual_output_hash": actual_hash, "template_hash": prepared.template_hash, "projection_receipt": prepared.projection_receipt, "path": relative_output(snapshot, view.output_path), "state": state, } @staticmethod def _config(snapshot: ProjectSnapshot) -> RenderConfig: if snapshot.descriptor.render is None: raise DocForgeError("render_not_configured", "Project has no configured render views") return snapshot.descriptor.render @staticmethod def _views(config: RenderConfig, view_id: str | None) -> tuple[RenderView, ...]: if view_id is None: return config.views views = tuple(view for view in config.views if view.view_id == view_id) if not views: raise DocForgeError( "unknown_render_view", "Render view is not declared by this project", view=view_id ) return views @staticmethod def _result(snapshot: ProjectSnapshot, **payload: object) -> dict[str, object]: return { "status": "ok", "project_id": snapshot.descriptor.project_id, "project_root_fingerprint": project_root_fingerprint(snapshot.descriptor.root), "adapter": snapshot.descriptor.adapter, "revision": snapshot.revision, "source_hash": snapshot.source_hash, **payload, } @contextmanager def _lock(self) -> Generator[None]: root = self.project.descriptor.cache_root if root.resolve(strict=False) != root: raise DocForgeError("path_escape", "Render lock directory is not safe") root.mkdir(parents=True, exist_ok=True) if not root.is_dir() or root.resolve(strict=False) != root: raise DocForgeError("path_escape", "Render lock directory is not safe") lock_path = root / "render.lock" try: descriptor = os.open(lock_path, os.O_RDWR | os.O_CREAT | os.O_NOFOLLOW, 0o600) except OSError as error: raise DocForgeError("path_escape", "Render lock path is not safe") from error with os.fdopen(descriptor, "a+b") as handle: fcntl.flock(handle.fileno(), fcntl.LOCK_EX) try: yield finally: fcntl.flock(handle.fileno(), fcntl.LOCK_UN)