diff --git a/README.md b/README.md index 5abe4d0..0d25d10 100644 --- a/README.md +++ b/README.md @@ -228,6 +228,30 @@ work_item/scripts/python -m pytest -m phase_f -q Phases A–E remain green in the full run above. +## Media and metadata hardening (US07-03) + +Every pixel the application reads goes through `photo_pipeline/imaging.py`: the +declared dimensions are checked before anything is decoded, Pillow's +decompression-bomb warning is treated as a refusal, JPEG decodes near the requested +size, and each decoder failure becomes one of two typed errors. A damaged file is a +per-item error with a persisted code, never a failed scan or a dead worker. + +Every metadata stage ends with an EXIF checkpoint (`services/exif_checkpoint.py`): +snapshot, write the owned keywords, read back, prove the owned fields landed and that +nothing else moved, refresh the file hash. A field the stage does not own that +changed anyway makes the checkpoint `divergent` — recorded in `exif_projections`, +shown in the review queue, never repaired behind the user's back, and not counted as +verified, so upload stays blocked. + +The golden corpus that proves all of it is generated, not committed: +`tests/fixtures/media_corpus.py` declares every format, orientation, profile, +damage, and metadata case with its expected outcome, and the suite regenerates it +twice to prove it does not drift. + +```bash +work_item/scripts/python -m pytest tests/integration/test_media_hardening.py tests/integration/test_exif_checkpoints.py -q +``` + ## Legacy CLI archive The command-line tools this application was extracted from are frozen in diff --git a/legacy_cli_archive/donor_ledger.yaml b/legacy_cli_archive/donor_ledger.yaml index 0044f17..b17deb9 100644 --- a/legacy_cli_archive/donor_ledger.yaml +++ b/legacy_cli_archive/donor_ledger.yaml @@ -480,6 +480,13 @@ rows: inference path needs the local model + deterministic fake. target: photo_pipeline/integrations/nsfw_model.py tests: [test_nsfwtag::test_score_images_cache_hit_skips_model] + delta: > + The donor set Pillow's process-global ImageFile.LOAD_TRUNCATED_IMAGES so a + partially downloaded file still scored. Here the same process also hashes + files and renders previews, and those must keep refusing a truncated file + rather than silently working on half of one; scoring opens images through + the bounded photo_pipeline.imaging door instead and skips the ones it cannot + read, leaving them unscored and visibly undecided (US07-03). status: characterized - id: nt-exif-keyword diff --git a/migrations/versions/0015_exif_projections.py b/migrations/versions/0015_exif_projections.py new file mode 100644 index 0000000..1b74aaa --- /dev/null +++ b/migrations/versions/0015_exif_projections.py @@ -0,0 +1,42 @@ +"""Durable EXIF projections per asset and stage (US07-03). + +Revision ID: 0015_exif_projections +Revises: 0014_restore_plans +Create Date: 2026-08-16 + +The concept's ``exif_projections`` table, added at the point it earns its keep: a +checkpoint that finds a field it does not own changed must be able to say so after +a restart. ``state`` is verified | divergent | failed, and only ``verified`` counts +as a completed metadata stage. +""" + +import sqlalchemy as sa +from alembic import op + +revision = "0015_exif_projections" +down_revision = "0014_restore_plans" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + op.create_table( + "exif_projections", + sa.Column("asset_id", sa.String(), sa.ForeignKey("assets.id"), primary_key=True), + sa.Column("stage", sa.String(), primary_key=True), # safety | analysis + sa.Column("id", sa.String(), nullable=False), + sa.Column("projection_version", sa.Integer(), nullable=False, server_default="0"), + # What the stage asked for: {"add": [...], "remove": [...]}. + sa.Column("desired_json", sa.String(), nullable=True), + # Fields outside the stage's ownership that did not survive the write. + sa.Column("divergent_fields", sa.String(), nullable=True), + sa.Column("result_file_sha256", sa.String(), nullable=True), + sa.Column("state", sa.String(), nullable=False), + sa.Column("error_code", sa.String(), nullable=True), + sa.Column("verified_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("updated_at", sa.DateTime(timezone=True), nullable=True), + ) + + +def downgrade() -> None: + op.drop_table("exif_projections") diff --git a/photo_pipeline/api/app.py b/photo_pipeline/api/app.py index 79ac1a7..97408f0 100644 --- a/photo_pipeline/api/app.py +++ b/photo_pipeline/api/app.py @@ -41,6 +41,7 @@ import photo_pipeline.jobs.domain_handlers # noqa: F401 from photo_pipeline.config import Config from photo_pipeline.db import create_db_engine, create_session_factory, run_migrations from photo_pipeline.logging import configure_logging +from photo_pipeline.services.thumbnails import ThumbnailService from photo_pipeline.services.upload_batches import UploadBatchService FRONTEND_DIR = Path(__file__).resolve().parents[2] / "frontend" @@ -96,6 +97,9 @@ def create_app(config: Config | None = None) -> FastAPI: # An upload whose process died left no outcome behind; resolve it now so the # uploader lane is free and the uncertain batch is visible (US05-02). UploadBatchService(app.state.session_factory, config=config).recover() + # A render killed mid-write leaves its temporary beside the cache entry; + # remove those recognized leftovers, and only those (US07-03). + ThumbnailService(app.state.session_factory, config).cleanup_temp_files() try: yield finally: diff --git a/photo_pipeline/imaging.py b/photo_pipeline/imaging.py new file mode 100644 index 0000000..6db1afc --- /dev/null +++ b/photo_pipeline/imaging.py @@ -0,0 +1,119 @@ +"""Bounded, defensive image decoding — one door for every pixel this app reads. + +A photo library contains files nobody planned for: truncated downloads, zero-byte +placeholders, a PNG whose header claims 200000×200000, a TIFF with a broken ICC +profile, an extension that lies about its content. None of them may take down a +request or a worker, and none may decode more pixels than the caller allowed +(concept §17: decoded pixels, not file size, are what exhausts memory). + +``open_image`` is that single door: + +* the declared dimensions are checked **before** a pixel is decoded; +* Pillow's decompression-bomb *warning* is promoted to an error, because the + warning band (between Pillow's limit and twice it) still decodes the image; +* every decoder failure — at open time or during the caller's decode — becomes one + of two typed errors, so callers map them to their own item state instead of + catching ``Exception``; +* error text names no path: it reaches API responses, and the full reason goes to + the server log instead (US07-02). + +``to_srgb`` and ``draft`` are the other two bounded-decode helpers: colour-manage a +profile-bearing image into sRGB, and let JPEG decode straight to a size near the +requested one rather than at full resolution. +""" + +from __future__ import annotations + +import io +import logging +import warnings +from contextlib import contextmanager +from pathlib import Path + +from PIL import Image, ImageCms, ImageFile, UnidentifiedImageError + +log = logging.getLogger(__name__) + +# Matches ``Config.thumbnail_max_pixels``; used where no configuration is at hand +# (hashing runs inside discovery, which takes no config). +DEFAULT_MAX_PIXELS = 100_000_000 + + +class MediaError(Exception): + """A file could not be turned into pixels safely.""" + + +class UndecodableImage(MediaError): + """Corrupt, truncated, empty, or not an image at all.""" + + +class ImageTooLarge(MediaError): + """More pixels than this operation is allowed to decode.""" + + +@contextmanager +def open_image(path: Path | str, *, max_pixels: int = DEFAULT_MAX_PIXELS): + """Yield an open :class:`PIL.Image.Image`, bounded and with typed failures. + + Decoder errors raised inside the ``with`` body are translated too — a truncated + JPEG only fails when its pixels are actually pulled, which is the caller's line, + not this one. + """ + # Pillow's truncation tolerance is a process-global switch that any library in + # the process can flip (the donor CLI did). This door decides the policy for its + # own callers: half a file is not a picture. + tolerated = ImageFile.LOAD_TRUNCATED_IMAGES + ImageFile.LOAD_TRUNCATED_IMAGES = False + with warnings.catch_warnings(): + # The warning band is not a warning for us: it means Pillow was willing to + # decode an image large enough to be a denial-of-service. + warnings.simplefilter("error", Image.DecompressionBombWarning) + try: + with Image.open(path) as image: + width, height = image.size + if width * height > max_pixels: + raise ImageTooLarge(f"{width}x{height} exceeds the {max_pixels} pixel limit") + yield image + except MediaError: + raise + except (Image.DecompressionBombError, Image.DecompressionBombWarning) as error: + log.info("refused oversized image %s: %s", path, error) + raise ImageTooLarge("image exceeds the decompression-bomb limit") from error + except (UnidentifiedImageError, OSError, ValueError, SyntaxError, MemoryError) as error: + log.info("cannot decode %s: %s", path, error) + raise UndecodableImage(f"cannot decode image ({type(error).__name__})") from error + finally: + ImageFile.LOAD_TRUNCATED_IMAGES = tolerated + + +def draft(image: Image.Image, size: int) -> None: + """Ask the decoder for a smaller image where the format allows it (JPEG). + + This is the difference between decoding a 40-megapixel JPEG and decoding the + roughly 1-megapixel version a 1280px preview needs. + """ + try: + image.draft(None, (size, size)) + except (AttributeError, ValueError, OSError): # not a draft-capable format + pass + + +def to_srgb(image: Image.Image, *, mode: str) -> Image.Image: + """Convert into ``mode``, colour-managing through an embedded ICC profile. + + Without this a wide-gamut original renders with visibly wrong colours, because + its numbers are interpreted as sRGB. A broken or unreadable profile is not a + reason to fail a preview — the plain conversion is still a correct picture. + """ + profile = image.info.get("icc_profile") + if profile: + try: + return ImageCms.profileToProfile( + image, + ImageCms.ImageCmsProfile(io.BytesIO(profile)), + ImageCms.createProfile("sRGB"), + outputMode=mode, + ) + except Exception as error: # noqa: BLE001 - any ICC failure falls back + log.info("ignoring unusable ICC profile on %s: %s", getattr(image, "filename", "?"), error) + return image.convert(mode) diff --git a/photo_pipeline/integrations/exiftool.py b/photo_pipeline/integrations/exiftool.py index 92cd64e..56aae86 100644 --- a/photo_pipeline/integrations/exiftool.py +++ b/photo_pipeline/integrations/exiftool.py @@ -56,6 +56,30 @@ def read_keyword_sets(paths: Iterable[str]) -> dict[str, set[str]]: return out +def read_all(path: str) -> dict | None: + """Every tag exiftool can read from ``path``, or ``None`` when it cannot answer. + + This is the snapshot an EXIF checkpoint compares against: proving that a write + preserved the fields it does not own requires knowing all of them, not just the + ones being written (US07-03). ``None`` (exiftool missing, unreadable file, + unparsable output) is not an empty snapshot — a caller must not read it as + "nothing was there". + """ + try: + result = subprocess.run( + ["exiftool", "-m", "-j", "-G0:1", path], capture_output=True, text=True + ) + except FileNotFoundError: + return None + try: + records = json.loads(result.stdout or "[]") + except ValueError: + return None + if not records: + return None + return {k: v for k, v in records[0].items() if k != "SourceFile"} + + def apply_keywords(path: str, *, add: Iterable[str] = (), remove: Iterable[str] = ()) -> bool: """Idempotently add/remove keywords in Keywords + Subject; preserve all else.""" args = ["exiftool", "-m", "-overwrite_original"] diff --git a/photo_pipeline/integrations/nsfw_model.py b/photo_pipeline/integrations/nsfw_model.py index b28327e..d2fbe36 100644 --- a/photo_pipeline/integrations/nsfw_model.py +++ b/photo_pipeline/integrations/nsfw_model.py @@ -15,6 +15,8 @@ from __future__ import annotations from pathlib import Path +from photo_pipeline import imaging + MODEL_ID = "AdamCodd/vit-base-nsfw-detector" BATCH = 16 @@ -54,13 +56,17 @@ class NsfwModel: self._ensure_loaded() import numpy as np import torch - from PIL import Image, ImageFile + from PIL import Image - ImageFile.LOAD_TRUNCATED_IMAGES = True - - def preprocess(image): - image = image.convert("RGB").resize((self._size, self._size), Image.BILINEAR) - array = (np.asarray(image, dtype="float32") / 255.0 - 0.5) / 0.5 + # The donor set ``ImageFile.LOAD_TRUNCATED_IMAGES = True`` here. That flag is + # process-global: in this application the same process also hashes files and + # renders previews, and those must keep failing loudly on a truncated file + # rather than quietly working on half of one (US07-03). An unreadable image + # is skipped instead — it stays unscored, and therefore visibly undecided. + def preprocess(path): + with imaging.open_image(path) as image: + small = image.convert("RGB").resize((self._size, self._size), Image.BILINEAR) + array = (np.asarray(small, dtype="float32") / 255.0 - 0.5) / 0.5 return torch.from_numpy(array).permute(2, 0, 1) results: list[tuple[str, float]] = [] @@ -69,9 +75,10 @@ class NsfwModel: tensors, batch_paths = [], [] for path in items[start : start + self.batch]: try: - tensors.append(preprocess(Image.open(path))) + tensors.append(preprocess(path)) batch_paths.append(path) - except Exception: + except (imaging.MediaError, OSError, ValueError): + # One bad file must not cost the batch its other fifteen. continue if not tensors: continue diff --git a/photo_pipeline/models/__init__.py b/photo_pipeline/models/__init__.py index 0beb416..a8234dd 100644 --- a/photo_pipeline/models/__init__.py +++ b/photo_pipeline/models/__init__.py @@ -7,6 +7,7 @@ Alembic environment relies on. from photo_pipeline.models.albums import AlbumProposal from photo_pipeline.models.archives import ArchiveLocation, ArchiveOperation, ArchivePlan from photo_pipeline.models.assets import Asset, AssetPath +from photo_pipeline.models.exif import ExifProjection from photo_pipeline.models.duplicates import ( DuplicateCluster, DuplicateMember, @@ -28,6 +29,7 @@ __all__ = [ "DuplicateCluster", "DuplicateMember", "DuplicateNegativeLink", + "ExifProjection", "Job", "JobItem", "JobEvent", diff --git a/photo_pipeline/models/exif.py b/photo_pipeline/models/exif.py new file mode 100644 index 0000000..96d9e0a --- /dev/null +++ b/photo_pipeline/models/exif.py @@ -0,0 +1,33 @@ +"""The durable EXIF projection per asset and stage (concept §3, US07-03). + +One row per ``(asset_id, stage)``: what the stage wanted written, what the file +looked like afterwards, and whether anything outside the stage's ownership moved. +``state = divergent`` is the whole point of the table — it survives restarts, keeps +the asset out of stages that require verified metadata, and gives a human something +to look at instead of a silent repair. +""" + +from __future__ import annotations + +from datetime import datetime + +from sqlalchemy import DateTime, ForeignKey, Integer, String +from sqlalchemy.orm import Mapped, mapped_column + +from photo_pipeline.db import Base + + +class ExifProjection(Base): + __tablename__ = "exif_projections" + + asset_id: Mapped[str] = mapped_column(ForeignKey("assets.id"), primary_key=True) + stage: Mapped[str] = mapped_column(String, primary_key=True) # safety | analysis + id: Mapped[str] = mapped_column(String, nullable=False) + projection_version: Mapped[int] = mapped_column(Integer, nullable=False, default=0) + desired_json: Mapped[str | None] = mapped_column(String) + divergent_fields: Mapped[str | None] = mapped_column(String) # JSON array + result_file_sha256: Mapped[str | None] = mapped_column(String) + state: Mapped[str] = mapped_column(String, nullable=False) # verified|divergent|failed + error_code: Mapped[str | None] = mapped_column(String) + verified_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) + updated_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) diff --git a/photo_pipeline/services/analysis.py b/photo_pipeline/services/analysis.py index 5c4ae8f..3698d5d 100644 --- a/photo_pipeline/services/analysis.py +++ b/photo_pipeline/services/analysis.py @@ -26,9 +26,8 @@ from sqlalchemy import func, select from sqlalchemy.orm import sessionmaker from photo_pipeline import path_policy -from photo_pipeline.integrations import exiftool from photo_pipeline.models import AnalysisResult, Asset, SafetyReview -from photo_pipeline.services import hashing +from photo_pipeline.services import exif_checkpoint from photo_pipeline.services.safety import SFW MODEL = "gemini-2.5-flash" @@ -170,7 +169,7 @@ class AnalysisService: tokens=int(result.get("_tokens", 0)) if isinstance(result, dict) else 0, raw=json.dumps(result, ensure_ascii=False), ) - _write_analysis_exif(path, result) + self._write_analysis_exif(asset_id, path, result) analyzed += 1 return {"analyzed": analyzed, "skipped": skipped, "errors": errors} @@ -195,10 +194,42 @@ class AnalysisService: row.location_hint = result.get("location_hint") row.approx_year = result.get("approx_year") row.analyzed_at = now - row.exif_written_at = now session.add(row) session.commit() + def _write_analysis_exif(self, asset_id: str, path: str, result: dict) -> None: + """The analysis EXIF checkpoint: additive keywords, then prove the rest held. + + Additive by design — safety keywords and the user's own keywords are merged + with, never replaced (concept §3). ``exif_written_at`` is set only when the + read-back verified both the new keywords and every field this stage does not + own; a divergent result is recorded and left for a human (US07-03). + ponytail: the managed ``AI:`` caption segment, once captions are owned here. + """ + tags = tuple(str(tag) for tag in (result.get("tags") or [])) + if not tags: + return + checkpoint = exif_checkpoint.run(path, add=tags) + exif_checkpoint.record( + self._session_factory, + asset_id=asset_id, + stage="analysis", + result=checkpoint, + add=tags, + ) + if not checkpoint.verified: + return + with self._session_factory() as session: + row = session.get(AnalysisResult, asset_id) + if row is not None: + row.exif_written_at = checkpoint.verified_at + asset = session.get(Asset, asset_id) + if asset is not None and checkpoint.sha256: + # The bytes changed when the container was rewritten; upload must use + # the hash of what is actually on disk now (concept §3). + asset.current_sha256 = checkpoint.sha256 + session.commit() + def get(self, asset_id: str) -> dict | None: with self._session_factory() as session: row = session.get(AnalysisResult, asset_id) @@ -211,17 +242,6 @@ def _album_hint(path: str) -> str: return Path(path).parent.name -def _write_analysis_exif(path: str, result: dict) -> None: - """Additive analysis keywords into EXIF (Keywords/Subject), preserving safety - and user keywords. The donor also wrote a managed caption; only keywords are - written here via the shared adapter. - ponytail: add the managed ``AI:`` caption segment + read-back preservation check - when the analysis EXIF checkpoint is hardened.""" - tags = result.get("tags") if isinstance(result, dict) else None - if tags: - exiftool.apply_keywords(path, add=[str(t) for t in tags]) - - def _result_dict(row: AnalysisResult) -> dict: data = {field: getattr(row, field) for field in RESULT_FIELDS} data["tags"] = json.loads(row.tags) if row.tags else [] diff --git a/photo_pipeline/services/exif_checkpoint.py b/photo_pipeline/services/exif_checkpoint.py new file mode 100644 index 0000000..17bf5b3 --- /dev/null +++ b/photo_pipeline/services/exif_checkpoint.py @@ -0,0 +1,173 @@ +"""The EXIF checkpoint every metadata stage ends with (concept §3, US07-03). + +A stage does not own a file's metadata; it owns a few fields in it. So writing is +never "set these tags" — it is: + + snapshot everything → write only the owned fields → read everything back + → prove the owned fields landed → prove nothing else moved + → refresh the file hash → record the projection + +Non-destructive here means *semantic* preservation: exiftool may rewrite the whole +container, so the file's bytes, size, and timestamps legitimately change. What may +not change is any field this stage does not own. When one does, the checkpoint is +``divergent``: the result is recorded, the stage is **not** marked verified, and +nothing is silently repaired — a later stage that needs verified metadata (upload) +therefore stays blocked until a human looks. + +``failed`` is the third outcome and is deliberately distinct: exiftool missing, an +unreadable file, or a write that did not take is not evidence that metadata is fine. +""" + +from __future__ import annotations + +import json +import uuid +from dataclasses import dataclass +from datetime import datetime, timezone + +from photo_pipeline.integrations import exiftool +from photo_pipeline.models import ExifProjection +from photo_pipeline.services import hashing + +# The tags the safety and analysis stages may change. Matched on the tag name, so +# every group carries the same rule (IPTC:Keywords, XMP:XMP-dc:Subject, ...). +OWNED_TAGS = frozenset({"Keywords", "Subject"}) + +# Not metadata about the picture: filesystem facts, the digest that necessarily +# moves whenever IPTC does, and the structural tags exiftool has to create the first +# time it writes an IPTC or XMP block. Comparing these would report every write as +# divergent and make the signal worthless. +VOLATILE_PREFIXES = ("File:System:", "ExifTool:") +VOLATILE_KEYS = frozenset( + { + "File:CurrentIPTCDigest", + "IPTC:ApplicationRecordVersion", + "XMP:XMP-x:XMPToolkit", + "XMP:XMP-xmp:MetadataDate", + } +) + +VERIFIED = "verified" +DIVERGENT = "divergent" +FAILED = "failed" + + +@dataclass(frozen=True) +class CheckpointResult: + state: str # verified | divergent | failed + changed_fields: tuple[str, ...] = () + sha256: str | None = None + verified_at: datetime | None = None + reason: str | None = None + + @property + def verified(self) -> bool: + return self.state == VERIFIED + + +def _now() -> datetime: + return datetime.now(timezone.utc) + + +def is_owned(key: str) -> bool: + return key.rsplit(":", 1)[-1] in OWNED_TAGS + + +def is_volatile(key: str) -> bool: + return key in VOLATILE_KEYS or key.startswith(VOLATILE_PREFIXES) + + +def compare(before: dict, after: dict) -> tuple[str, ...]: + """Fields outside this stage's ownership whose value did not survive the write. + + Additions count: a tag that appears out of nowhere is as much a divergence as a + tag that disappeared — both mean the write did more than it was asked to. + """ + keys = set(before) | set(after) + return tuple( + sorted( + key + for key in keys + if not is_owned(key) + and not is_volatile(key) + and before.get(key) != after.get(key) + ) + ) + + +def owned_values(snapshot: dict) -> set[str]: + """Lowercased Keywords/Subject values across every group in the snapshot.""" + values: set[str] = set() + for key, value in snapshot.items(): + if not is_owned(key): + continue + items = value if isinstance(value, list) else [value] + values.update(str(item).strip().lower() for item in items if item is not None) + return values + + +def run( + path: str, + *, + add: tuple[str, ...] = (), + remove: tuple[str, ...] = (), +) -> CheckpointResult: + """Write the owned keywords for one asset and verify the whole file around them.""" + before = exiftool.read_all(path) + if before is None: + return CheckpointResult(FAILED, reason="metadata_unreadable") + + if not exiftool.apply_keywords(path, add=add, remove=remove): + return CheckpointResult(FAILED, reason="write_failed") + + after = exiftool.read_all(path) + if after is None: + return CheckpointResult(FAILED, reason="readback_unreadable") + + present = owned_values(after) + wanted = {value.strip().lower() for value in add} + unwanted = {value.strip().lower() for value in remove} + if not wanted <= present or (unwanted & present): + return CheckpointResult(FAILED, reason="owned_fields_not_written") + + changed = compare(before, after) + sha256 = hashing.sha256_file(path) + if changed: + return CheckpointResult(DIVERGENT, changed_fields=changed, sha256=sha256) + return CheckpointResult(VERIFIED, sha256=sha256, verified_at=_now()) + + +def record( + session_factory, + *, + asset_id: str, + stage: str, + result: CheckpointResult, + add: tuple[str, ...] = (), + remove: tuple[str, ...] = (), +) -> None: + """Persist the projection for ``(asset_id, stage)`` — one current row per pair. + + The row is what makes divergence durable and reviewable rather than a log line + that scrolled away. + """ + with session_factory() as session: + row = session.get(ExifProjection, (asset_id, stage)) + if row is None: + row = ExifProjection(asset_id=asset_id, stage=stage, id=str(uuid.uuid4())) + session.add(row) + row.projection_version = (row.projection_version or 0) + 1 + row.desired_json = json.dumps({"add": list(add), "remove": list(remove)}) + row.divergent_fields = json.dumps(list(result.changed_fields)) + row.result_file_sha256 = result.sha256 + row.state = result.state + row.error_code = result.reason + row.verified_at = result.verified_at + row.updated_at = _now() + session.commit() + + +def state_for(session_factory, asset_id: str, stage: str) -> str | None: + with session_factory() as session: + row = session.get(ExifProjection, (asset_id, stage)) + return row.state if row else None diff --git a/photo_pipeline/services/hashing.py b/photo_pipeline/services/hashing.py index 9718683..6fb5dea 100644 --- a/photo_pipeline/services/hashing.py +++ b/photo_pipeline/services/hashing.py @@ -16,7 +16,9 @@ from __future__ import annotations import hashlib from pathlib import Path -from PIL import Image, ImageOps +from PIL import ImageOps + +from photo_pipeline import imaging PIXEL_HASH_VERSION = 1 PHASH_VERSION = 1 @@ -41,7 +43,9 @@ def _digest_file(path: Path | str, digest) -> str: def pixel_sha256(path: Path | str) -> str: - with Image.open(path) as image: + # Bounded decode: a header claiming a billion pixels is refused before it is + # ever turned into memory (US07-03). + with imaging.open_image(path) as image: oriented = ImageOps.exif_transpose(image) rgb = oriented.convert("RGB") header = f"{PIXEL_HASH_VERSION}:{rgb.width}x{rgb.height}:".encode() @@ -68,7 +72,9 @@ def phash(path: Path | str) -> str: import numpy as np from scipy.fftpack import dct - with Image.open(path) as image: + from PIL import Image + + with imaging.open_image(path) as image: small = image.convert("L").resize((32, 32), Image.LANCZOS) matrix = np.asarray(small, dtype=np.float64) transformed = dct(dct(matrix, axis=0), axis=1) diff --git a/photo_pipeline/services/safety.py b/photo_pipeline/services/safety.py index 5983c45..00652da 100644 --- a/photo_pipeline/services/safety.py +++ b/photo_pipeline/services/safety.py @@ -108,9 +108,8 @@ from datetime import datetime, timezone from sqlalchemy import select from sqlalchemy.orm import sessionmaker -from photo_pipeline.integrations import exiftool -from photo_pipeline.models import Asset, SafetyReview -from photo_pipeline.services import hashing +from photo_pipeline.models import Asset, ExifProjection, SafetyReview +from photo_pipeline.services import exif_checkpoint DECISIONS = {SFW, NSFW, "deferred"} @@ -174,6 +173,14 @@ class SafetyService: with self._session_factory() as session: assets = list(session.scalars(_eligible_assets_query().order_by(Asset.current_path))) latest = self._latest_by_asset(session) + # One query, not one per asset: the reviewer needs to see a divergent + # checkpoint, which is neither "verified" nor a plain failure (US07-03). + projections = { + row.asset_id: row.state + for row in session.scalars( + select(ExifProjection).where(ExifProjection.stage == "safety") + ) + } rows = [] for asset in assets: review = latest.get(asset.id) @@ -189,6 +196,7 @@ class SafetyService: "decision": decision, "suggested": classify(review.score) if review and review.score is not None else None, "exif_verified": bool(review and review.exif_verified_at), + "exif_state": projections.get(asset.id), } ) return {"total": len(rows), "items": rows[offset : offset + limit]} @@ -247,13 +255,24 @@ class SafetyService: result_sha256 = None if write_exif and decision in (SFW, NSFW) and path: ops = exif_projection(decision) - if exiftool.apply_keywords(path, add=ops["add"], remove=ops["remove"]): - # Read back: the chosen keyword present, the opposite absent. - keywords = exiftool.read_keyword_sets([path]).get(path, set()) - opposite = NSFW if decision == SFW else SFW - if decision in keywords and opposite not in keywords: - exif_verified_at = _now() - result_sha256 = hashing.sha256_file(path) + # The full checkpoint: write the owned keyword, read the whole file back, + # and prove every field this stage does not own survived. A divergent + # result is recorded and left alone — it must not count as verified, so + # upload stays blocked until a human decides (US07-03). + result = exif_checkpoint.run( + path, add=tuple(ops["add"]), remove=tuple(ops["remove"]) + ) + exif_checkpoint.record( + self._session_factory, + asset_id=asset_id, + stage="safety", + result=result, + add=tuple(ops["add"]), + remove=tuple(ops["remove"]), + ) + if result.verified: + exif_verified_at = result.verified_at + result_sha256 = result.sha256 now = _now() with self._session_factory() as session: diff --git a/photo_pipeline/services/thumbnails.py b/photo_pipeline/services/thumbnails.py index fb7fce9..a2a3a29 100644 --- a/photo_pipeline/services/thumbnails.py +++ b/photo_pipeline/services/thumbnails.py @@ -20,15 +20,16 @@ pa-imaging). from __future__ import annotations +import os import uuid from pathlib import Path -from PIL import Image, ImageOps, UnidentifiedImageError +from PIL import Image, ImageOps from sqlalchemy import select from sqlalchemy.exc import IntegrityError from sqlalchemy.orm import sessionmaker -from photo_pipeline import path_policy +from photo_pipeline import imaging, path_policy from photo_pipeline.config import Config from photo_pipeline.models import Asset, Thumbnail from photo_pipeline.services import availability @@ -42,6 +43,9 @@ except Exception: # pragma: no cover pass SIZES = (256, 512, 1280) +# Every in-flight render writes one of these beside its destination; startup +# cleanup recognises exactly this pattern and nothing else. +TEMP_SUFFIX = ".tmp" THUMB_VERSION = 1 THUMB_FORMAT = "webp" # The size kept as durable comparison evidence for archived assets (concept §9). @@ -222,28 +226,33 @@ class ThumbnailService: return self._cache_dir / safe[:2] / f"{safe}.{THUMB_FORMAT}" def _render(self, source: str, size: int, cache_key: str) -> dict: + """Decode bounded, orient, colour-manage, resize, and write atomically. + + The temporary file is removed on every failure path: a decoder that dies + halfway through ``save`` would otherwise leave a stray ``.tmp`` in the cache + forever (US07-03). + """ + destination = self._cache_path(cache_key) + destination.parent.mkdir(parents=True, exist_ok=True) + tmp = destination.with_name(f".{destination.name}.{uuid.uuid4().hex}{TEMP_SUFFIX}") try: - with Image.open(source) as image: - width, height = image.size - if width * height > self._config.thumbnail_max_pixels: - raise ImageTooLarge( - f"{width}x{height} exceeds {self._config.thumbnail_max_pixels} px" - ) + with imaging.open_image(source, max_pixels=self._config.thumbnail_max_pixels) as image: + imaging.draft(image, size) # JPEG decodes near the target size oriented = ImageOps.exif_transpose(image) mode = "RGBA" if _has_alpha(oriented) else "RGB" - converted = oriented.convert(mode) + converted = imaging.to_srgb(oriented, mode=mode) converted.thumbnail((size, size), Image.LANCZOS) out_width, out_height = converted.size - destination = self._cache_path(cache_key) - destination.parent.mkdir(parents=True, exist_ok=True) - tmp = destination.with_name(f".{destination.name}.{uuid.uuid4().hex}.tmp") converted.save(tmp, format="WEBP", quality=82, method=4) - except ImageTooLarge: + except imaging.ImageTooLarge as error: + tmp.unlink(missing_ok=True) + raise ImageTooLarge(str(error)) from None + except imaging.UndecodableImage as error: + tmp.unlink(missing_ok=True) + raise UnsupportedImage(str(error)) from None + except Exception: + tmp.unlink(missing_ok=True) raise - except (UnidentifiedImageError, OSError, ValueError) as error: - raise UnsupportedImage(f"cannot decode {source}: {error}") from error - - import os os.replace(tmp, destination) return { @@ -253,6 +262,28 @@ class ThumbnailService: "format": THUMB_FORMAT, } + def cleanup_temp_files(self) -> int: + """Remove leftover render temporaries, and only those. + + Runs at startup, where the concept is explicit: clean *recognized* stale + temporary files, never arbitrary ones. The pattern is this service's own — + a dot-prefixed name inside the managed cache directory ending in + ``.tmp`` — so nothing outside the cache and nothing a user put there can + match. Returns how many were removed. + """ + if not self._cache_dir.is_dir(): + return 0 + removed = 0 + for leftover in self._cache_dir.rglob(f".*{TEMP_SUFFIX}"): + if not leftover.is_file() or leftover.is_symlink(): + continue + try: + leftover.unlink() + removed += 1 + except OSError: # pragma: no cover - a racing render already won + pass + return removed + # ── persistence ──────────────────────────────────────────────────────────── def _record_ready( self, cache_key: str, asset_id: str, size: int, rendered: dict, *, protected: bool = False diff --git a/tests/fixtures/media_corpus.py b/tests/fixtures/media_corpus.py new file mode 100644 index 0000000..7a7f235 --- /dev/null +++ b/tests/fixtures/media_corpus.py @@ -0,0 +1,325 @@ +"""The golden media corpus: every format, orientation, profile, damage, and +metadata case this application claims to survive (US07-03, concept §18). + +``CASES`` is the manifest and the authority. Each entry declares a stable logical +id (never a path), how the file is generated, and what the pipeline must do with +it — decode it, or refuse it with one precise error code. Tests parametrize over the +manifest, so a case that is added here without an expectation, or an expectation +that stops holding, fails the suite rather than quietly going untested. + +Everything is generated, never committed: fixed pixel seeds, fixed EXIF strings, no +clock, no network, no personal data. Regeneration is byte-stable, which +``test_media_hardening.py`` proves by building the corpus twice and comparing +checksums — a golden corpus that drifts is not golden. +""" + +from __future__ import annotations + +import struct +import zlib +from dataclasses import dataclass, field +from pathlib import Path +from typing import Callable + +import numpy as np +from PIL import Image + +# Error codes the pipeline may answer with; ``None`` means "must render". +UNSUPPORTED = "unsupported_image" +TOO_LARGE = "image_too_large" + +# One fixed capture time for every metadata case: the corpus must not depend on when +# it was generated. +CAPTURE_TIME = "2019:07:14 10:30:00" + + +@dataclass(frozen=True) +class MediaCase: + id: str + filename: str + kind: str # format | orientation | profile | damaged | metadata + build: Callable[[Path], None] + expect_error: str | None = None + # Declared for the metadata cases: exiftool arguments applied after the pixels + # are written, and the user fields that must survive every later stage. + exif_args: tuple[str, ...] = () + preserved_fields: tuple[str, ...] = () + notes: str = "" + tags: tuple[str, ...] = field(default_factory=tuple) + + +# ── generators ──────────────────────────────────────────────────────────────── + + +def _pixels(width: int, height: int, seed: int, bands: int = 3) -> np.ndarray: + return np.random.default_rng(seed).integers(0, 256, (height, width, bands), dtype=np.uint8) + + +def _save(path: Path, image: Image.Image, **kwargs) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + image.save(path, **kwargs) + + +def _jpeg(width=320, height=240, seed=1, **kwargs): + def build(path: Path) -> None: + _save(path, Image.fromarray(_pixels(width, height, seed)), quality=90, **kwargs) + + return build + + +def _oriented_jpeg(orientation: int): + def build(path: Path) -> None: + image = Image.fromarray(_pixels(400, 200, 5)) # landscape source + exif = image.getexif() + exif[274] = orientation # 0x0112 Orientation + _save(path, image, exif=exif, quality=90) + + return build + + +def _rotated_pixels(path: Path) -> None: + """The same scene rotated in the pixels instead of in a tag.""" + _save(path, Image.fromarray(_pixels(400, 200, 5)).transpose(Image.ROTATE_90), quality=90) + + +def _png(alpha: bool = False, seed: int = 2): + def build(path: Path) -> None: + if alpha: + _save(path, Image.fromarray(_pixels(120, 90, seed, bands=4), "RGBA")) + else: + _save(path, Image.fromarray(_pixels(120, 90, seed))) + + return build + + +def _webp(path: Path) -> None: + _save(path, Image.fromarray(_pixels(150, 100, 3)), quality=80) + + +def _tiff(path: Path) -> None: + _save(path, Image.fromarray(_pixels(140, 110, 4))) + + +def _grayscale(path: Path) -> None: + _save(path, Image.fromarray(_pixels(100, 100, 6)).convert("L"), quality=90) + + +def _cmyk(path: Path) -> None: + _save(path, Image.fromarray(_pixels(100, 100, 7)).convert("CMYK"), quality=90) + + +def _tiny(path: Path) -> None: + _save(path, Image.fromarray(_pixels(1, 1, 8))) + + +def _icc_tagged(path: Path) -> None: + """A profile-bearing image: the colour-managed decode path must run.""" + from PIL import ImageCms + + profile = bytearray(ImageCms.ImageCmsProfile(ImageCms.createProfile("sRGB")).tobytes()) + # An ICC header carries its creation timestamp at bytes 24..35. Left alone, the + # corpus would be a different corpus every time it is generated. + profile[24:36] = b"\x00" * 12 + _save(path, Image.fromarray(_pixels(120, 80, 9)), icc_profile=bytes(profile), quality=90) + + +def _broken_icc(path: Path) -> None: + """A profile that is not a profile: a picture is still a picture.""" + _save(path, Image.fromarray(_pixels(120, 80, 10)), icc_profile=b"not-a-profile", quality=90) + + +def _wide_jpeg(path: Path) -> None: + """Large enough that decoding it at full resolution is visible in memory.""" + _save(path, Image.fromarray(_pixels(4000, 3000, 11)), quality=70) + + +def _zero_byte(path: Path) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_bytes(b"") + + +def _truncated_jpeg(path: Path) -> None: + image = Image.fromarray(_pixels(400, 300, 12)) + _save(path, image, quality=90) + data = path.read_bytes() + path.write_bytes(data[: len(data) // 2]) # header intact, pixels missing + + +def _corrupt_png(path: Path) -> None: + image = Image.fromarray(_pixels(120, 90, 13)) + _save(path, image) + data = bytearray(path.read_bytes()) + data[40:80] = b"\x00" * 40 # shred the compressed stream, keep the header + path.write_bytes(bytes(data)) + + +def _not_an_image(path: Path) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_bytes(b"This is a text file that happens to be named .jpg\n") + + +def _png_declaring(width: int, height: int): + """A tiny, structurally valid PNG whose header claims an enormous picture. + + A few hundred bytes on disk, gigapixels on paper: the pipeline must refuse it + from the declared dimensions, before a single pixel is allocated. Rewriting the + IHDR of a real PNG (rather than hand-rolling a stub) keeps the file openable, so + the refusal is proven to come from the size check and not from a parse failure. + """ + + def build(path: Path) -> None: + import io + + buffer = io.BytesIO() + Image.fromarray(_pixels(4, 4, 15)).save(buffer, format="PNG") + data = bytearray(buffer.getvalue()) + start = 8 + 4 # PNG signature, then the IHDR length field + struct.pack_into(">II", data, start + 4, width, height) + ihdr = bytes(data[start : start + 4 + 13]) + struct.pack_into(">I", data, start + 4 + 13, zlib.crc32(ihdr)) + path.parent.mkdir(parents=True, exist_ok=True) + path.write_bytes(bytes(data)) + + return build + + +CASES: tuple[MediaCase, ...] = ( + # ── formats ─────────────────────────────────────────────────────────────── + MediaCase("jpeg", "formats/plain.jpg", "format", _jpeg()), + MediaCase("jpeg_uppercase_ext", "formats/UPPER.JPG", "format", _jpeg(seed=14)), + MediaCase("png", "formats/plain.png", "format", _png()), + MediaCase("png_alpha", "formats/alpha.png", "format", _png(alpha=True)), + MediaCase("webp", "formats/plain.webp", "format", _webp), + MediaCase("tiff", "formats/plain.tiff", "format", _tiff), + # ── orientation ─────────────────────────────────────────────────────────── + *( + MediaCase( + f"orientation_{value}", + f"orientation/exif_{value}.jpg", + "orientation", + _oriented_jpeg(value), + notes="EXIF orientation must be applied before resizing", + ) + for value in range(1, 9) + ), + MediaCase("rotated_pixels", "orientation/rotated.jpg", "orientation", _rotated_pixels), + # ── colour and size profiles ────────────────────────────────────────────── + MediaCase("grayscale", "profiles/gray.jpg", "profile", _grayscale), + MediaCase("cmyk", "profiles/cmyk.jpg", "profile", _cmyk), + MediaCase("tiny", "profiles/tiny.png", "profile", _tiny), + MediaCase("icc_tagged", "profiles/icc.jpg", "profile", _icc_tagged), + MediaCase( + "broken_icc", + "profiles/broken-icc.jpg", + "profile", + _broken_icc, + notes="an unusable ICC profile falls back to a plain conversion, never an error", + ), + MediaCase( + "large_jpeg", + "profiles/large.jpg", + "profile", + _wide_jpeg, + notes="12 megapixels: the decode must stay near the requested size", + ), + # ── damaged and hostile inputs ──────────────────────────────────────────── + MediaCase("zero_byte", "damaged/empty.jpg", "damaged", _zero_byte, UNSUPPORTED), + MediaCase("truncated_jpeg", "damaged/truncated.jpg", "damaged", _truncated_jpeg, UNSUPPORTED), + MediaCase("corrupt_png", "damaged/corrupt.png", "damaged", _corrupt_png, UNSUPPORTED), + MediaCase("text_as_jpeg", "damaged/text.jpg", "damaged", _not_an_image, UNSUPPORTED), + MediaCase( + "bomb_header", + "damaged/bomb.png", + "damaged", + _png_declaring(60_000, 60_000), + TOO_LARGE, + notes="3.6 gigapixels declared in the header and nothing else", + ), + MediaCase( + "bomb_warning_band", + "damaged/bomb-warning.png", + "damaged", + _png_declaring(10_000, 10_000), + TOO_LARGE, + notes="inside Pillow's warn-only band; the warning is promoted to a refusal", + ), + # ── metadata ────────────────────────────────────────────────────────────── + MediaCase("no_exif", "metadata/bare.jpg", "metadata", _jpeg(seed=20)), + MediaCase( + "user_exif", + "metadata/user.jpg", + "metadata", + _jpeg(seed=21), + exif_args=( + "-Artist=Ada Lovelace", + "-Copyright=(c) Ada", + f"-DateTimeOriginal={CAPTURE_TIME}", + "-GPSLatitude=48.137", + "-GPSLatitudeRef=N", + "-Rating=4", + "-ImageDescription=A day out", + ), + preserved_fields=( + "EXIF:IFD0:Artist", + "EXIF:IFD0:Copyright", + "EXIF:ExifIFD:DateTimeOriginal", + "EXIF:IFD0:ImageDescription", + "XMP:XMP-xmp:Rating", + ), + notes="user metadata that every stage must leave exactly as it found it", + ), + MediaCase( + "prior_safety_keyword", + "metadata/prior-safety.jpg", + "metadata", + _jpeg(seed=22), + exif_args=("-Keywords+=nsfw", "-Subject+=nsfw", "-Artist=Ada Lovelace"), + preserved_fields=("EXIF:IFD0:Artist",), + notes="a safety decision already written by an earlier run", + ), + MediaCase( + "prior_analysis_keywords", + "metadata/prior-analysis.jpg", + "metadata", + _jpeg(seed=23), + exif_args=("-Keywords+=beach", "-Keywords+=sunset", "-Subject+=beach", "-Subject+=sunset"), + notes="analysis keywords from an earlier run; a safety write must not drop them", + ), + MediaCase( + "conflicting_safety_keywords", + "metadata/conflicting.jpg", + "metadata", + _jpeg(seed=24), + exif_args=("-Keywords+=sfw", "-Keywords+=nsfw", "-Subject+=sfw", "-Subject+=nsfw"), + notes="both safety keywords at once: mutually exclusive means one must go", + ), + MediaCase( + "malformed_metadata", + "metadata/malformed.jpg", + "metadata", + _jpeg(seed=25, exif=b"\x00\x01\x02not-a-valid-exif-block"), + notes="a broken EXIF block must not stop the picture from being usable", + ), +) + +CASES_BY_ID = {case.id: case for case in CASES} + + +def build_corpus(root: Path, *, ids: tuple[str, ...] | None = None) -> dict[str, Path]: + """Generate the corpus (or a named subset) under ``root``; return id → path.""" + import subprocess + + built: dict[str, Path] = {} + for case in CASES: + if ids is not None and case.id not in ids: + continue + path = root / case.filename + case.build(path) + if case.exif_args: + subprocess.run( + ["exiftool", "-m", "-overwrite_original", *case.exif_args, str(path)], + capture_output=True, + check=False, + ) + built[case.id] = path + return built diff --git a/tests/integration/test_exif_checkpoints.py b/tests/integration/test_exif_checkpoints.py new file mode 100644 index 0000000..b1426bb --- /dev/null +++ b/tests/integration/test_exif_checkpoints.py @@ -0,0 +1,297 @@ +"""US07-03: EXIF checkpoints, asserted with before/after metadata snapshots. + +Every test here reads the complete metadata of a real file through exiftool before +the stage runs and again afterwards, then compares the two. That is the only way to +prove the property the concept actually asks for: a stage owns a few fields and must +leave literally everything else — dates, GPS, camera, artist, rating, other people's +keywords — exactly as it found them. + +The other half is divergence. When something outside the stage's ownership does move, +the checkpoint must say so, refuse to call itself verified, and change nothing back: +a silent repair is how a library quietly loses the user's metadata. +""" + +from __future__ import annotations + +import json +import shutil +import subprocess +import uuid +from datetime import datetime, timezone +from types import SimpleNamespace + +import pytest + +from photo_pipeline.config import Config +from photo_pipeline.db import create_db_engine, create_session_factory, run_migrations +from photo_pipeline.integrations import exiftool +from photo_pipeline.models import Asset, ExifProjection +from photo_pipeline.services import exif_checkpoint +from photo_pipeline.services.analysis import AnalysisService +from photo_pipeline.services.safety import SafetyService +from tests.fixtures.media_corpus import CASES_BY_ID, build_corpus + +pytestmark = pytest.mark.skipif( + shutil.which("exiftool") is None, reason="exiftool not installed" +) + + +@pytest.fixture +def env(tmp_path): + data = tmp_path / "data" + data.mkdir() + lib = tmp_path / "lib" + files = build_corpus( + lib, + ids=( + "user_exif", + "prior_safety_keyword", + "prior_analysis_keywords", + "conflicting_safety_keywords", + "malformed_metadata", + "no_exif", + ), + ) + config = Config.from_env( + {"PHOTO_PIPELINE_DATA_DIR": str(data), "PHOTO_PIPELINE_LIBRARY_ROOTS": str(lib)} + ) + run_migrations(config.database_url) + engine = create_db_engine(config.database_url) + sf = create_session_factory(engine) + assets = {} + with sf() as session: + for case_id, path in files.items(): + asset = Asset( + id=str(uuid.uuid4()), + original_path=str(path), + current_path=str(path), + discovered_at=datetime.now(timezone.utc), + hash_version=1, + ) + session.add(asset) + assets[case_id] = asset.id + session.commit() + yield SimpleNamespace(config=config, lib=lib, sf=sf, files=files, assets=assets) + engine.dispose() + + +class Provider: + """A vision provider whose tags are fixed, so the EXIF assertion is the test.""" + + def __init__(self, tags): + self.tags = list(tags) + + def analyze(self, path, *, album_hint): + return {"description": "a photo", "tags": self.tags} + + +def snapshot(path): + return exiftool.read_all(str(path)) + + +def stage_fields(before, after): + """Everything that changed except this application's owned and volatile tags.""" + return exif_checkpoint.compare(before, after) + + +# ── preservation ────────────────────────────────────────────────────────────── + + +def test_a_safety_decision_preserves_every_user_field(env): + case = CASES_BY_ID["user_exif"] + path = env.files["user_exif"] + before = snapshot(path) + assert before, "the fixture must actually carry user metadata" + + result = SafetyService(env.sf).decide(env.assets["user_exif"], "nsfw") + after = snapshot(path) + + assert result["exif_verified"] is True + assert stage_fields(before, after) == () + for field in case.preserved_fields: + assert after[field] == before[field], field + assert "nsfw" in exif_checkpoint.owned_values(after) + assert "sfw" not in exif_checkpoint.owned_values(after) + + +def test_analysis_keywords_are_additive_and_keep_the_safety_decision(env): + """The two stages share the Keywords field; the second must merge, not replace.""" + asset_id = env.assets["prior_analysis_keywords"] + path = env.files["prior_analysis_keywords"] + SafetyService(env.sf).decide(asset_id, "sfw") + before = snapshot(path) + + AnalysisService( + env.sf, provider=Provider(["harbour", "boats"]), library_roots=(env.lib,) + ).run([asset_id]) + after = snapshot(path) + + keywords = exif_checkpoint.owned_values(after) + assert {"sfw", "beach", "sunset", "harbour", "boats"} <= keywords + assert "nsfw" not in keywords + assert stage_fields(before, after) == () + + +def test_flipping_a_safety_decision_removes_only_the_opposite_keyword(env): + asset_id = env.assets["prior_analysis_keywords"] + path = env.files["prior_analysis_keywords"] + safety = SafetyService(env.sf) + safety.decide(asset_id, "nsfw") + before = snapshot(path) + + safety.decide(asset_id, "sfw") + after = snapshot(path) + + keywords = exif_checkpoint.owned_values(after) + assert "sfw" in keywords and "nsfw" not in keywords + assert {"beach", "sunset"} <= keywords, "analysis keywords are not safety's to remove" + assert stage_fields(before, after) == () + + +def test_conflicting_safety_keywords_are_resolved_to_one(env): + asset_id = env.assets["conflicting_safety_keywords"] + SafetyService(env.sf).decide(asset_id, "sfw") + keywords = exif_checkpoint.owned_values(snapshot(env.files["conflicting_safety_keywords"])) + assert keywords & {"sfw", "nsfw"} == {"sfw"} + + +def test_malformed_metadata_does_not_block_the_checkpoint(env): + result = SafetyService(env.sf).decide(env.assets["malformed_metadata"], "nsfw") + assert result["exif_verified"] is True + + +# ── verification and the refreshed hash ─────────────────────────────────────── + + +def test_a_verified_checkpoint_refreshes_the_recorded_bytes(env): + """exiftool rewrites the container, so the stored SHA-256 must be the new one — + upload compares against exactly these bytes.""" + asset_id = env.assets["no_exif"] + path = env.files["no_exif"] + SafetyService(env.sf).decide(asset_id, "sfw") + + with env.sf() as session: + asset = session.get(Asset, asset_id) + row = session.get(ExifProjection, (asset_id, "safety")) + import hashlib + + on_disk = hashlib.sha256(path.read_bytes()).hexdigest() + assert asset.current_sha256 == on_disk + assert row.state == "verified" and row.result_file_sha256 == on_disk + assert json.loads(row.desired_json) == {"add": ["sfw"], "remove": ["nsfw"]} + + +def test_the_projection_survives_a_restart(env): + asset_id = env.assets["user_exif"] + SafetyService(env.sf).decide(asset_id, "nsfw") + + engine = create_db_engine(env.config.database_url) # a fresh connection, as a restart is + try: + with create_session_factory(engine)() as session: + row = session.get(ExifProjection, (asset_id, "safety")) + assert row.state == "verified" and row.verified_at is not None + finally: + engine.dispose() + + +# ── divergence ──────────────────────────────────────────────────────────────── + + +def test_a_field_the_stage_does_not_own_changing_is_divergent(env, monkeypatch): + """Something rewrote the artist while the safety keyword was being written. + + The stage must not call that verified, must record what moved, and must not put + the old value back — the file is now a question for a human. + """ + asset_id = env.assets["user_exif"] + path = env.files["user_exif"] + real_apply = exiftool.apply_keywords + + def sabotage(target, *, add=(), remove=()): + ok = real_apply(target, add=add, remove=remove) + subprocess.run( + ["exiftool", "-m", "-overwrite_original", "-Artist=Someone Else", str(target)], + capture_output=True, + check=False, + ) + return ok + + monkeypatch.setattr(exiftool, "apply_keywords", sabotage) + result = SafetyService(env.sf).decide(asset_id, "nsfw") + + assert result["exif_verified"] is False, "a divergent checkpoint is not verified" + with env.sf() as session: + row = session.get(ExifProjection, (asset_id, "safety")) + asset = session.get(Asset, asset_id) + assert row.state == "divergent" + assert "EXIF:IFD0:Artist" in json.loads(row.divergent_fields) + assert row.verified_at is None + # Not repaired, and not silently accepted as the current verified bytes. + assert snapshot(path)["EXIF:IFD0:Artist"] == "Someone Else" + assert asset.current_sha256 is None + + +def test_a_divergent_asset_shows_up_in_the_review_queue(env, monkeypatch): + asset_id = env.assets["user_exif"] + monkeypatch.setattr( + exif_checkpoint, + "run", + lambda *args, **kwargs: exif_checkpoint.CheckpointResult( + exif_checkpoint.DIVERGENT, changed_fields=("EXIF:IFD0:Artist",), sha256="abc" + ), + ) + SafetyService(env.sf).decide(asset_id, "sfw") + + rows = SafetyService(env.sf).review_queue()["items"] + row = next(item for item in rows if item["asset_id"] == asset_id) + assert row["exif_state"] == "divergent" + assert row["exif_verified"] is False + + +def test_a_write_that_does_not_take_is_a_failure_not_a_verification(env, monkeypatch): + monkeypatch.setattr(exiftool, "apply_keywords", lambda *a, **k: False) + result = SafetyService(env.sf).decide(env.assets["no_exif"], "sfw") + assert result["exif_verified"] is False + assert exif_checkpoint.state_for(env.sf, env.assets["no_exif"], "safety") == "failed" + + +def test_unreadable_metadata_is_a_failure_not_an_empty_snapshot(env, monkeypatch): + """``None`` from exiftool means "cannot answer"; treating it as "nothing there" + would make every field look preserved.""" + monkeypatch.setattr(exiftool, "read_all", lambda path: None) + result = exif_checkpoint.run(str(env.files["no_exif"]), add=("sfw",)) + assert result.state == "failed" and result.reason == "metadata_unreadable" + + +def test_analysis_records_its_own_projection_separately(env): + asset_id = env.assets["user_exif"] + SafetyService(env.sf).decide(asset_id, "sfw") + AnalysisService(env.sf, provider=Provider(["pier"]), library_roots=(env.lib,)).run([asset_id]) + + assert exif_checkpoint.state_for(env.sf, asset_id, "safety") == "verified" + assert exif_checkpoint.state_for(env.sf, asset_id, "analysis") == "verified" + + +# ── the comparison rules themselves ─────────────────────────────────────────── + + +def test_compare_ignores_owned_and_volatile_fields_only(): + before = { + "EXIF:IFD0:Artist": "Ada", + "IPTC:Keywords": ["holiday"], + "File:System:FileSize": "3.8 kB", + "File:CurrentIPTCDigest": "aaa", + "XMP:XMP-x:XMPToolkit": "old", + } + after = { + "EXIF:IFD0:Artist": "Ada", + "IPTC:Keywords": ["holiday", "sfw"], + "File:System:FileSize": "3.9 kB", + "File:CurrentIPTCDigest": "bbb", + "XMP:XMP-x:XMPToolkit": "new", + } + assert exif_checkpoint.compare(before, after) == () + + after["EXIF:GPS:GPSLatitude"] = "48.1" # an addition counts as much as a loss + del after["EXIF:IFD0:Artist"] + assert exif_checkpoint.compare(before, after) == ("EXIF:GPS:GPSLatitude", "EXIF:IFD0:Artist") diff --git a/tests/integration/test_media_hardening.py b/tests/integration/test_media_hardening.py new file mode 100644 index 0000000..22bc474 --- /dev/null +++ b/tests/integration/test_media_hardening.py @@ -0,0 +1,335 @@ +"""US07-03: the golden media corpus, driven through the real decode path. + +Every case in ``tests/fixtures/media_corpus.py`` is exercised here. The claims: + +* a supported format/orientation/profile renders, with the orientation applied and + the decode bounded to roughly the size that was asked for; +* a damaged, empty, lying, or gigapixel file becomes one precise item error — the + scan still finishes, the other assets still render, and a job that meets one keeps + running; +* a failure leaves no temporary file behind, and startup removes only the temporaries + this service recognises; +* cache invalidation follows the pixels, not the metadata. +""" + +from __future__ import annotations + +import hashlib +import shutil +import time +import tracemalloc +from types import SimpleNamespace + +import pytest +from PIL import Image + +from photo_pipeline import imaging +from photo_pipeline.config import Config +from photo_pipeline.db import create_db_engine, create_session_factory, run_migrations +from photo_pipeline.models import Asset, Thumbnail +from photo_pipeline.services.inventory import InventoryService +from photo_pipeline.services.thumbnails import ( + TEMP_SUFFIX, + ThumbnailError, + ThumbnailService, +) +from tests.fixtures.media_corpus import CASES, CASES_BY_ID, build_corpus + +RENDERABLE = [case for case in CASES if case.expect_error is None] +DAMAGED = [case for case in CASES if case.expect_error is not None] + +needs_exiftool = pytest.mark.skipif( + shutil.which("exiftool") is None, reason="exiftool not installed" +) + + +@pytest.fixture(scope="module") +def corpus(tmp_path_factory): + root = tmp_path_factory.mktemp("corpus") + return SimpleNamespace(root=root, files=build_corpus(root)) + + +@pytest.fixture +def env(tmp_path, corpus): + """A library holding the whole corpus, scanned into a fresh database.""" + data = tmp_path / "data" + data.mkdir() + lib = tmp_path / "lib" + shutil.copytree(corpus.root, lib) + config = Config.from_env( + {"PHOTO_PIPELINE_DATA_DIR": str(data), "PHOTO_PIPELINE_LIBRARY_ROOTS": str(lib)} + ) + run_migrations(config.database_url) + engine = create_db_engine(config.database_url) + sf = create_session_factory(engine) + scan = InventoryService(sf).scan(lib) + by_id = { + case.id: next( + (aid for path, aid in scan.asset_ids.items() if path.endswith(case.filename)), None + ) + for case in CASES + } + yield SimpleNamespace( + config=config, lib=lib, sf=sf, scan=scan, assets=by_id, + thumbs=ThumbnailService(sf, config), + ) + engine.dispose() + + +# ── the manifest itself ─────────────────────────────────────────────────────── + + +def test_the_manifest_is_internally_consistent(): + assert len({case.id for case in CASES}) == len(CASES), "logical ids must be unique" + assert len({case.filename for case in CASES}) == len(CASES), "paths must be unique" + for case in CASES: + assert case.kind in {"format", "orientation", "profile", "damaged", "metadata"} + assert case.expect_error in (None, "unsupported_image", "image_too_large") + + +def test_the_corpus_regenerates_byte_for_byte(tmp_path): + """A golden corpus that drifts between runs cannot be a golden corpus.""" + first = build_corpus(tmp_path / "one") + second = build_corpus(tmp_path / "two") + digests = { + case_id: ( + hashlib.sha256(first[case_id].read_bytes()).hexdigest(), + hashlib.sha256(second[case_id].read_bytes()).hexdigest(), + ) + for case_id in first + } + drifted = [case_id for case_id, (a, b) in digests.items() if a != b] + assert not drifted, f"non-deterministic fixtures: {drifted}" + + +# ── bounded decode ──────────────────────────────────────────────────────────── + + +@pytest.mark.parametrize("case", RENDERABLE, ids=lambda case: case.id) +def test_every_supported_case_renders(case, env): + asset_id = env.assets[case.id] + assert asset_id, f"{case.id} was not discovered by the scan" + path = env.thumbs.generate(asset_id, 256) + with Image.open(path) as thumb: + assert thumb.format == "WEBP" + assert max(thumb.size) <= 256 + assert min(thumb.size) >= 1 + + +@pytest.mark.parametrize("orientation", range(1, 9)) +def test_exif_orientation_is_applied_before_resizing(orientation, env): + """All eight tags: the 90° ones must come out portrait from a landscape source.""" + asset_id = env.assets[f"orientation_{orientation}"] + with Image.open(env.thumbs.generate(asset_id, 256)) as thumb: + rotated = orientation in (5, 6, 7, 8) + assert (thumb.height > thumb.width) is rotated + + +def test_transparency_and_grayscale_survive_the_pipeline(env): + with Image.open(env.thumbs.generate(env.assets["png_alpha"], 256)) as thumb: + assert "A" in thumb.getbands() + with Image.open(env.thumbs.generate(env.assets["grayscale"], 256)) as thumb: + assert thumb.size == (100, 100) # smaller than the request: never upscaled + + +def test_a_broken_colour_profile_still_produces_a_picture(env): + """An unusable ICC profile is a metadata problem, not a reason to lose the preview.""" + for case_id in ("icc_tagged", "broken_icc"): + with Image.open(env.thumbs.generate(env.assets[case_id], 256)) as thumb: + assert thumb.size[0] > 0 + + +def test_a_large_jpeg_is_not_decoded_at_full_resolution(env): + """12 megapixels would be ~36 MB of pixels; the draft decode keeps it far below.""" + tracemalloc.start() + try: + env.thumbs.generate(env.assets["large_jpeg"], 256) + _, peak = tracemalloc.get_traced_memory() + finally: + tracemalloc.stop() + assert peak < 12_000_000, f"decode peaked at {peak} bytes" + + +# ── damaged inputs ──────────────────────────────────────────────────────────── + + +@pytest.mark.parametrize("case", DAMAGED, ids=lambda case: case.id) +def test_every_damaged_case_is_a_precise_item_error(case, env): + asset_id = env.assets[case.id] + assert asset_id, f"{case.id} was not discovered by the scan" + with pytest.raises(ThumbnailError) as raised: + env.thumbs.generate(asset_id, 256) + assert raised.value.code == case.expect_error + # Persisted, so a broken original is not re-decoded on every request... + with env.sf() as session: + rows = [ + row + for row in session.query(Thumbnail).all() + if row.asset_id == asset_id and row.state == "error" + ] + assert rows and rows[0].error_code == case.expect_error + # ...and the cached failure is the same precise error, not a generic one. + with pytest.raises(ThumbnailError) as again: + env.thumbs.generate(asset_id, 256) + assert again.value.code == case.expect_error + + +def test_a_gigapixel_header_is_refused_quickly_and_cheaply(env): + """The refusal must come from the declared size, not from decoding it.""" + tracemalloc.start() + started = time.monotonic() + try: + with pytest.raises(ThumbnailError): + env.thumbs.generate(env.assets["bomb_header"], 1280) + _, peak = tracemalloc.get_traced_memory() + finally: + tracemalloc.stop() + assert time.monotonic() - started < 5 + assert peak < 5_000_000, f"a 3.6 gigapixel header allocated {peak} bytes" + + +def test_one_broken_file_stops_neither_the_scan_nor_its_neighbours(env): + """The whole corpus is in the library, damaged files included.""" + assert len(env.scan.asset_ids) >= len(CASES) - 1 # the empty file has no pixels to hash + for case in RENDERABLE[:5]: + assert env.thumbs.generate(env.assets[case.id], 256).exists() + + +def test_a_job_that_meets_a_broken_file_gets_evidence_not_an_exception(env): + """``ensure_protected`` is what the archive lane calls; a plan must not die on + one unreadable original.""" + evidence = env.thumbs.ensure_protected(env.assets["corrupt_png"]) + assert evidence["state"] == "unsupported" + assert evidence["error_code"] == "unsupported_image" + assert env.thumbs.ensure_protected(env.assets["jpeg"])["state"] == "ready" + + +def test_undecodable_files_do_not_break_hashing(env): + """Discovery records what it can: byte identity always, pixel identity when the + file has pixels.""" + with env.sf() as session: + assets = {a.current_path: a for a in session.query(Asset).all()} + for case in DAMAGED: + asset = next((a for path, a in assets.items() if path.endswith(case.filename)), None) + if asset is None: + continue + assert asset.current_sha256, "byte identity is always available" + assert asset.pixel_sha256 is None, "undecodable files must not invent pixel identity" + + +# ── cache lifecycle ─────────────────────────────────────────────────────────── + + +def test_a_failed_render_leaves_no_temporary_behind(env): + with pytest.raises(ThumbnailError): + env.thumbs.generate(env.assets["truncated_jpeg"], 256) + leftovers = list(env.config.thumbnail_cache_dir.rglob(f"*{TEMP_SUFFIX}")) + assert leftovers == [] + + +def test_startup_cleanup_removes_only_recognised_temporaries(env): + cache = env.config.thumbnail_cache_dir + env.thumbs.generate(env.assets["jpeg"], 256) # creates the cache directory + stale = cache / f".abc123{TEMP_SUFFIX}" + stale.write_bytes(b"half a thumbnail") + innocent = cache / "keep-me.webp" + innocent.write_bytes(b"not a temporary") + + assert env.thumbs.cleanup_temp_files() == 1 + assert not stale.exists() + assert innocent.exists() + assert list(cache.rglob("*.webp")), "real cache entries survive" + + +def test_metadata_only_change_reuses_the_thumbnail_and_a_pixel_change_does_not(env): + """The cache key follows the pixels: an EXIF edit must not cost a re-render.""" + asset_id = env.assets["jpeg"] + first = env.thumbs.generate(asset_id, 256) + + with env.sf() as session: # what a rescan records after an EXIF-only edit + asset = session.get(Asset, asset_id) + asset.current_sha256 = "different-bytes" + session.commit() + assert env.thumbs.generate(asset_id, 256) == first + + with env.sf() as session: # a genuine pixel change + asset = session.get(Asset, asset_id) + asset.pixel_sha256 = "different-pixels" + session.commit() + assert env.thumbs.generate(asset_id, 256) != first + + +def test_a_deleted_cache_file_is_regenerated(env): + asset_id = env.assets["png"] + path = env.thumbs.generate(asset_id, 256) + path.unlink() + regenerated = env.thumbs.generate(asset_id, 256) + assert regenerated == path and regenerated.exists() + + +# ── the imaging door itself ─────────────────────────────────────────────────── + + +def test_open_image_translates_every_decoder_failure(corpus): + """Callers must be able to catch two typed errors, never bare ``Exception``.""" + for case in DAMAGED: + expected = imaging.ImageTooLarge if case.expect_error == "image_too_large" else imaging.UndecodableImage + with pytest.raises(expected): + with imaging.open_image(corpus.files[case.id]) as image: + image.load() + + +def test_open_image_refuses_more_pixels_than_the_caller_allowed(corpus): + with pytest.raises(imaging.ImageTooLarge): + with imaging.open_image(corpus.files["jpeg"], max_pixels=100): + pass + with imaging.open_image(corpus.files["jpeg"], max_pixels=100_000) as image: + assert image.size == (320, 240) + + +@needs_exiftool +def test_malformed_metadata_does_not_stop_the_picture(env): + """A broken EXIF block is a metadata fact, not a decode failure.""" + assert env.thumbs.generate(env.assets["malformed_metadata"], 256).exists() + with env.sf() as session: + asset = session.get(Asset, env.assets["malformed_metadata"]) + assert asset.pixel_sha256, "pixels are still identifiable" + + +def test_scoring_never_relaxes_truncated_image_handling(corpus): + """The donor's process-global ``LOAD_TRUNCATED_IMAGES`` is gone for good. + + It is global state: switching it on for the safety model would also switch it on + for hashing and preview rendering in the same process, and half a file would + silently become a valid picture (donor_ledger: nt-score-model). + """ + import inspect + + from PIL import ImageFile + + from photo_pipeline.integrations import nsfw_model + + code = [ + line + for line in inspect.getsource(nsfw_model).splitlines() + if not line.strip().startswith("#") + ] + assert not any("LOAD_TRUNCATED_IMAGES" in line for line in code) + + # And the door holds even when something else in the process turned it on — + # the frozen donor does exactly that when the characterization suite imports it. + previous = ImageFile.LOAD_TRUNCATED_IMAGES + ImageFile.LOAD_TRUNCATED_IMAGES = True + try: + with pytest.raises(imaging.UndecodableImage): + with imaging.open_image(corpus.files["truncated_jpeg"]) as image: + image.load() + assert ImageFile.LOAD_TRUNCATED_IMAGES is True, "the caller's setting is restored" + finally: + ImageFile.LOAD_TRUNCATED_IMAGES = previous + + +def test_every_manifest_case_is_exercised(): + """The corpus lint: no fixture may sit in the manifest untested.""" + covered = {case.id for case in RENDERABLE} | {case.id for case in DAMAGED} + assert covered == set(CASES_BY_ID) diff --git a/tests/story_traceability.json b/tests/story_traceability.json index 8a1c3d3..cd1f570 100644 --- a/tests/story_traceability.json +++ b/tests/story_traceability.json @@ -152,6 +152,10 @@ "tests/unit/test_security_policy.py", "tests/integration/test_security_boundaries.py", "tests/e2e/test_security.py" + ], + "US07-03": [ + "tests/integration/test_media_hardening.py", + "tests/integration/test_exif_checkpoints.py" ] } }