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")
|
