from __future__ import annotations from dataclasses import dataclass import hashlib import os from pathlib import Path import re from typing import Callable from .models import ContractError, ErrorCode from .process import ProcessSupervisor @dataclass(frozen=True) class PdfEvidence: path: Path bytes: int sha256: str pdf_magic_valid: bool openable: bool page_count: int | None encrypted: bool | None @dataclass(frozen=True) class PublishResult: final_path: Path duplicate: bool evidence: PdfEvidence def sha256_file(path: Path, chunk_size: int = 1 << 20, checkpoint: Callable[[], None] | None = None) -> str: digest = hashlib.sha256() with path.open("rb") as stream: while chunk := stream.read(chunk_size): if checkpoint: checkpoint() digest.update(chunk) if checkpoint: checkpoint() return digest.hexdigest() def validate_pdf(path: Path, *, pdfinfo_executable: Path | None = None, supervisor: ProcessSupervisor | None = None, timeout_ms: int = 20_000, checkpoint: Callable[[], None] | None = None) -> PdfEvidence: if checkpoint: checkpoint() if not path.is_file() or path.is_symlink(): raise ContractError(ErrorCode.PDF_NOT_OPENABLE, "path", "ordinary file required") size = path.stat().st_size with path.open("rb") as stream: magic = stream.read(5) == b"%PDF-" if not magic: raise ContractError(ErrorCode.PDF_MAGIC_INVALID, "path", "missing %PDF-") digest = sha256_file(path, checkpoint=checkpoint) pages = None encrypted = None openable = True if pdfinfo_executable: sup = supervisor or ProcessSupervisor() result = sup.run((str(pdfinfo_executable), str(path)), timeout_ms=timeout_ms) if result.error_code or result.exit_code != 0: raise ContractError(ErrorCode.PDF_NOT_OPENABLE, "pdfinfo", "nonzero/timeout") text = result.stdout_bytes.decode("utf-8", "replace") match = re.search(r"(?m)^Pages:\s+(\d+)\s*$", text) if not match: raise ContractError(ErrorCode.PAGE_COUNT_MISMATCH, "pdfinfo", "Pages absent") pages = int(match.group(1)) enc = re.search(r"(?mi)^Encrypted:\s+(yes|no)", text) encrypted = None if not enc else enc.group(1).lower() == "yes" if encrypted: raise ContractError(ErrorCode.PDF_ENCRYPTED, "pdfinfo", "encrypted") if checkpoint: checkpoint() return PdfEvidence(path, size, digest, True, openable, pages, encrypted) def publish_no_replace(staging: Path, final_path: Path, *, expected_sha256: str, pdfinfo_executable: Path | None = None, supervisor: ProcessSupervisor | None = None, timeout_ms: int = 20_000, checkpoint: Callable[[], None] | None = None) -> PublishResult: staging_evidence = validate_pdf(staging, pdfinfo_executable=pdfinfo_executable, supervisor=supervisor, timeout_ms=timeout_ms, checkpoint=checkpoint) if staging_evidence.sha256 != expected_sha256: raise ContractError(ErrorCode.HASH_MISMATCH, "staging", "expected hash mismatch") final_path.parent.mkdir(parents=True, exist_ok=True) if final_path.exists() or final_path.is_symlink(): if final_path.is_file() and not final_path.is_symlink(): existing = validate_pdf(final_path, pdfinfo_executable=pdfinfo_executable, supervisor=supervisor, timeout_ms=timeout_ms, checkpoint=checkpoint) if existing.sha256 == expected_sha256: return PublishResult(final_path, True, existing) raise ContractError(ErrorCode.FINAL_PATH_CONFLICT, "final_path", "existing object differs") # Copy into an O_EXCL staging file in the final directory first. The final # hard-link publication is therefore same-volume and remains no-replace even # when the run staging directory is on another volume. temp_path = final_path.parent / f".{final_path.name}.{expected_sha256[:16]}.hibor-staging" temp_created = False try: flags = os.O_CREAT | os.O_EXCL | os.O_WRONLY | getattr(os, "O_BINARY", 0) try: fd = os.open(temp_path, flags, 0o600) temp_created = True try: with staging.open("rb") as source: while chunk := source.read(1 << 20): if checkpoint: checkpoint() view = memoryview(chunk) while view: written = os.write(fd, view) if written <= 0: raise OSError("short write") view = view[written:] os.fsync(fd) finally: os.close(fd) except FileExistsError: if not temp_path.is_file() or temp_path.is_symlink(): raise ContractError(ErrorCode.FINAL_PATH_CONFLICT, "publish_staging", "existing object") temp_evidence = validate_pdf(temp_path, pdfinfo_executable=pdfinfo_executable, supervisor=supervisor, timeout_ms=timeout_ms, checkpoint=checkpoint) if temp_evidence.sha256 != expected_sha256 or temp_evidence.bytes != staging_evidence.bytes: raise ContractError(ErrorCode.FINAL_PATH_CONFLICT, "publish_staging", "recovery mismatch") temp_evidence = validate_pdf(temp_path, pdfinfo_executable=pdfinfo_executable, supervisor=supervisor, timeout_ms=timeout_ms, checkpoint=checkpoint) if temp_evidence.sha256 != expected_sha256 or temp_evidence.bytes != staging_evidence.bytes: raise ContractError(ErrorCode.HASH_MISMATCH, "publish_staging", "copy mismatch") os.link(temp_path, final_path) except FileExistsError: raise ContractError(ErrorCode.FINAL_PATH_CONFLICT, "final_path", "race") except OSError as exc: raise ContractError(ErrorCode.PUBLISH_FAILED, "final_path", str(exc)) from exc finally: # A pre-existing recovery object is not owned by this attempt. Never # unlink it: the creator may still be publishing or may need it after a # crash. Only the attempt that won O_EXCL owns cleanup. if temp_created: try: temp_path.unlink() except OSError: pass evidence = validate_pdf(final_path, pdfinfo_executable=pdfinfo_executable, supervisor=supervisor, timeout_ms=timeout_ms, checkpoint=checkpoint) if evidence.sha256 != expected_sha256 or evidence.bytes != staging_evidence.bytes: raise ContractError(ErrorCode.HASH_MISMATCH, "final_path", "post-publish mismatch") try: staging.unlink() except OSError: pass return PublishResult(final_path, False, evidence) def create_exclusive_bytes(path: Path, data: bytes, checkpoint: Callable[[], None] | None = None) -> None: path.parent.mkdir(parents=True, exist_ok=True) flags = os.O_CREAT | os.O_EXCL | os.O_WRONLY | getattr(os, "O_BINARY", 0) fd = os.open(path, flags, 0o600) try: view = memoryview(data) while view: if checkpoint: checkpoint() written = os.write(fd, view) if written <= 0: raise OSError("short write") view = view[written:] os.fsync(fd) finally: os.close(fd) if checkpoint: checkpoint() if path.read_bytes() != data: raise ContractError(ErrorCode.PERSIST_LATE, "path", "readback mismatch")