US07-03: Harden Media and Metadata Edge Cases

README.md

@@ -17,6 +17,25 @@ python -m photo_pipeline serve     # start the API + static review UI (127.0.0.1
 Configuration comes from `PHOTO_PIPELINE_*` environment variables (see
 `photo_pipeline/config.py`); secrets are referenced, never logged.
 
+### API access (US07-02)
+
+The app listens on loopback, so its attacker is another page in the same browser.
+Every `/api/v1` route except `health/live`, `health/ready`, and `session` requires
+the application session, and every mutation requires its CSRF token as well:
+
+```bash
+BASE=http://127.0.0.1:8000
+TOKEN=$(curl -sc /tmp/pp.jar $BASE/api/v1/session | python -c 'import json,sys; print(json.load(sys.stdin)["csrf_token"])')
+curl -sb /tmp/pp.jar -H "X-CSRF-Token: $TOKEN" -X POST $BASE/api/v1/albums/proposals -d '{}' -H 'Content-Type: application/json'
+```
+
+The session is per server process — restarting `serve` invalidates it, and the
+browser client re-bootstraps by itself. Requests are also refused when the `Host` is
+not a loopback name (DNS rebinding), when `Origin` is any other origin, when
+`Sec-Fetch-Site` says the request came from another site (an `<img>` pointed at a
+thumbnail), or when the body exceeds `PHOTO_PIPELINE_MAX_REQUEST_BYTES`. There is no
+CORS middleware at all, so no other origin can read a response.
+
 ## Testing
 
 One offline command runs the whole suite (unit, integration, and browser
@@ -209,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

frontend/js/api.js

@@ -2,14 +2,43 @@
 // cancellation. Every method accepts an optional { signal } from cancellable().
 export const BASE = "/api/v1";
 
+// The API refuses every request without the session cookie, and every mutation
+// without this token echoed back. The token is readable only same-origin, which is
+// what makes it proof that the caller is this app and not another page.
+let csrfToken = null;
+
+async function session() {
+  if (csrfToken === null) {
+    const response = await fetch(BASE + "/session", { credentials: "same-origin" });
+    const body = await response.json().catch(() => null);
+    csrfToken = (body && body.csrf_token) || null;
+  }
+  return csrfToken || "";
+}
+
+async function send(path, { signal, ...options }) {
+  return fetch(BASE + path, {
+    credentials: "same-origin",
+    signal,
+    ...options,
+    headers: {
+      "Content-Type": "application/json",
+      "X-CSRF-Token": await session(),
+      ...(options.headers || {}),
+    },
+  });
+}
+
 async function request(path, { signal, ...options } = {}) {
   let response;
   try {
-    response = await fetch(BASE + path, {
-      headers: { "Content-Type": "application/json" },
-      signal,
-      ...options,
-    });
+    response = await send(path, { signal, ...options });
+    // A restarted server issues a new session; re-bootstrap once rather than
+    // stranding an open tab on 401.
+    if (response.status === 401) {
+      csrfToken = null;
+      response = await send(path, { signal, ...options });
+    }
   } catch (error) {
     // A caller-cancelled fetch is not a failure; tag it so views can ignore it.
     if (error.name === "AbortError") {

frontend/js/tests/unit.js

@@ -29,6 +29,10 @@ function jsonResponse(status, body) {
 const tick = (ms = 0) => new Promise((r) => setTimeout(r, ms));
 
 async function run() {
+  // The client fetches its CSRF token once, lazily (US07-02). Do that against the
+  // real server first, so the stubbed fetch below only ever sees the call under test.
+  await api.workflow().catch(() => {});
+
   // ── store ────────────────────────────────────────────────────────────────
   {
     const store = createStore({ n: 0 });

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

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

photo_pipeline/api/app.py

@@ -8,11 +8,15 @@ and exposes the versioned ``/api/v1`` surface; US01-02 ships only health.
 
 from __future__ import annotations
 
+import logging
 from contextlib import asynccontextmanager
 from pathlib import Path
 
-from fastapi import FastAPI
+from fastapi import FastAPI, Request
+from fastapi.exceptions import RequestValidationError
+from fastapi.responses import JSONResponse
 from fastapi.staticfiles import StaticFiles
+from starlette.exceptions import HTTPException as StarletteHTTPException
 
 from photo_pipeline.api.routes import (
     albums,
@@ -25,20 +29,58 @@ from photo_pipeline.api.routes import (
     library,
     renames,
     safety,
+    session as session_routes,
     thumbnails,
     uploads,
     workflow,
 )
+from photo_pipeline.api.security import DEFAULT_HEADERS, SecurityMiddleware, Session
 
 # Registers the safety_score / analysis job handlers on import.
 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"
 
+log = logging.getLogger(__name__)
+
+
+def _envelope(status: int, code: str, message: str) -> JSONResponse:
+    return JSONResponse(
+        status_code=status,
+        content={"error": {"code": code, "message": message}},
+        headers=DEFAULT_HEADERS,
+    )
+
+
+def _install_error_handlers(app: FastAPI) -> None:
+    """One JSON error envelope everywhere, and nothing behind it.
+
+    An unhandled exception carries the library's absolute paths, SQL, and sometimes
+    a credential in its text; the client gets a code, the operator gets the traceback
+    in the server log (US07-02).
+    """
+
+    @app.exception_handler(StarletteHTTPException)
+    async def _http_error(request: Request, exc: StarletteHTTPException):
+        return _envelope(exc.status_code, "http_error", str(exc.detail))
+
+    @app.exception_handler(RequestValidationError)
+    async def _validation_error(request: Request, exc: RequestValidationError):
+        # Field locations only: the echoed input can be the caller's own data, but it
+        # is also what ends up in shared logs and screenshots.
+        fields = sorted(".".join(str(part) for part in error["loc"]) for error in exc.errors())
+        return _envelope(422, "invalid_request", f"invalid request fields: {', '.join(fields)}")
+
+    @app.exception_handler(Exception)
+    async def _unhandled(request: Request, exc: Exception):
+        log.exception("unhandled error serving %s", request.url.path)
+        return _envelope(500, "internal_error", "internal error")
+
 
 def create_app(config: Config | None = None) -> FastAPI:
     config = config or Config.from_env()
@@ -55,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:
@@ -62,6 +107,12 @@ def create_app(config: Config | None = None) -> FastAPI:
             app.state.engine = None
 
     app = FastAPI(title="Photo Pipeline", version="0.1.0", lifespan=lifespan)
+    # One session per process: the browser exchanges it for a cookie + CSRF token,
+    # and every other origin is refused before a route ever runs (US07-02).
+    app.state.session = Session.create()
+    app.add_middleware(SecurityMiddleware, session=app.state.session, config=config)
+    _install_error_handlers(app)
+    app.include_router(session_routes.router, prefix="/api/v1")
     app.include_router(health.router, prefix="/api/v1")
     app.include_router(inventory.router, prefix="/api/v1")
     app.include_router(duplicates.router, prefix="/api/v1")

photo_pipeline/api/routes/analysis.py

@@ -18,7 +18,10 @@ router = APIRouter(tags=["analysis"])
 
 
 def _service(request: Request) -> AnalysisService:
-    return AnalysisService(request.app.state.session_factory)
+    return AnalysisService(
+        request.app.state.session_factory,
+        library_roots=tuple(request.app.state.config.library_roots),
+    )
 
 
 def _error(status: int, code: str, message: str) -> JSONResponse:

photo_pipeline/api/routes/session.py

@@ -0,0 +1,30 @@
+"""Session bootstrap: the one endpoint reachable without a session (US07-02).
+
+It sets the ``HttpOnly``/``SameSite=Strict`` session cookie and returns the CSRF
+token in the body. A foreign page can call this — it just cannot read the answer,
+because the app sends no CORS headers — and the cookie it received is never attached
+to a request that foreign page initiates.
+"""
+
+from __future__ import annotations
+
+from fastapi import APIRouter, Request
+from fastapi.responses import JSONResponse
+
+from photo_pipeline.api.security import SESSION_COOKIE
+
+router = APIRouter(tags=["session"])
+
+
+@router.get("/session")
+def start_session(request: Request) -> JSONResponse:
+    session = request.app.state.session
+    response = JSONResponse({"csrf_token": session.csrf_token})
+    response.set_cookie(
+        SESSION_COOKIE,
+        session.id,
+        httponly=True,
+        samesite="strict",
+        path="/",
+    )
+    return response

photo_pipeline/api/routes/thumbnails.py

@@ -31,5 +31,7 @@ def get_thumbnail(asset_id: str, request: Request, size: int = Query(512)):
     return FileResponse(
         path,
         media_type="image/webp",
-        headers={"Cache-Control": "public, max-age=31536000, immutable"},
+        # private: the URL is versioned and immutable, but these bytes are the user's
+        # photos and must never sit in a shared cache (US07-02).
+        headers={"Cache-Control": "private, max-age=31536000, immutable"},
     )

photo_pipeline/api/security.py

@@ -0,0 +1,192 @@
+"""Local-web attack surface: session, CSRF, Origin/Host checks, default headers.
+
+The app binds to 127.0.0.1, so the attacker is not a remote client but another page
+in the user's browser (concept §15, "Local web attack"): any site can issue requests
+to ``http://127.0.0.1:8000`` and can embed ``<img src=...>`` against media endpoints.
+The defenses stack, because each one alone has a hole:
+
+* **Host** must be a loopback name — a DNS rebinding host that resolves to 127.0.0.1
+  passes the browser's origin rules but not this check.
+* **Origin**, when the browser sends one, must be this exact origin (scheme, host,
+  port). There is no CORS middleware at all, so a foreign page can never *read* a
+  response even if it manages to send a request.
+* **Sec-Fetch-Site** rejects cross-site loads that carry no Origin, which is what an
+  ``<img>`` or ``<script>`` against a media endpoint looks like.
+* A **session cookie** (``SameSite=Strict``, ``HttpOnly``) is required by every
+  ``/api/v1`` route except liveness/readiness and the bootstrap itself. Strict means
+  the browser never attaches it to a request another site initiated.
+* A **CSRF token** must be echoed in a header on every mutation. It is handed out
+  only in the bootstrap response body, which a foreign page cannot read (no CORS) —
+  so possessing it proves the caller is same-origin.
+
+``evaluate`` is a pure function over the request metadata: the whole policy is one
+table that a unit test can enumerate, and the middleware only applies its verdict.
+"""
+
+from __future__ import annotations
+
+import secrets
+from collections.abc import Mapping
+from dataclasses import dataclass
+from urllib.parse import urlsplit
+
+from starlette.datastructures import Headers, MutableHeaders
+from starlette.responses import JSONResponse
+
+SESSION_COOKIE = "pp_session"
+CSRF_HEADER = "x-csrf-token"
+API_PREFIX = "/api/v1"
+SAFE_METHODS = frozenset({"GET", "HEAD", "OPTIONS"})
+# Reachable without a session: liveness/readiness (an orchestrator has no cookie)
+# and the bootstrap that issues the session in the first place.
+PUBLIC_PATHS = frozenset(
+    {f"{API_PREFIX}/health/live", f"{API_PREFIX}/health/ready", f"{API_PREFIX}/session"}
+)
+LOOPBACK_HOSTS = frozenset({"127.0.0.1", "localhost", "::1", "[::1]"})
+
+# Applied to every response. No inline script/style is used by the frontend, so the
+# policy can stay strict; `frame-ancestors 'none'` and CORP keep other pages from
+# embedding the app or its thumbnails.
+DEFAULT_HEADERS = {
+    "x-content-type-options": "nosniff",
+    "x-frame-options": "DENY",
+    "referrer-policy": "no-referrer",
+    "cross-origin-resource-policy": "same-origin",
+    "cross-origin-opener-policy": "same-origin",
+    "content-security-policy": (
+        "default-src 'self'; img-src 'self' data:; style-src 'self'; script-src 'self'; "
+        "connect-src 'self'; frame-ancestors 'none'; base-uri 'none'; form-action 'none'"
+    ),
+}
+
+
+@dataclass(frozen=True)
+class Session:
+    """One process, one session. A local app has exactly one user; a session store
+    would be bookkeeping without a second subject to distinguish.
+
+    ponytail: per-session rows if the app ever serves more than one operator.
+    """
+
+    id: str
+    csrf_token: str
+
+    @classmethod
+    def create(cls) -> Session:
+        return cls(secrets.token_urlsafe(32), secrets.token_urlsafe(32))
+
+
+@dataclass(frozen=True)
+class Refusal:
+    status: int
+    code: str
+    message: str
+
+
+def split_host(value: str) -> tuple[str, str]:
+    """``"127.0.0.1:8000"`` -> ``("127.0.0.1", "8000")``; bracketed IPv6 aware."""
+    value = value.strip()
+    if value.startswith("["):
+        host, _, port = value.partition("]")
+        return host + "]", port.lstrip(":")
+    host, _, port = value.partition(":")
+    return host, port
+
+
+def evaluate(
+    *,
+    method: str,
+    path: str,
+    headers: Mapping[str, str],
+    session: Session,
+    allowed_hosts: frozenset[str] = LOOPBACK_HOSTS,
+    max_request_bytes: int,
+) -> Refusal | None:
+    """Why this request must be refused, or ``None`` when it may proceed."""
+    host_header = headers.get("host", "")
+    host, port = split_host(host_header)
+    if host.lower() not in allowed_hosts:
+        return Refusal(403, "host_not_allowed", "request host is not a local address")
+
+    origin = headers.get("origin")
+    if origin is not None and origin != "":
+        parts = urlsplit(origin)
+        origin_host, origin_port = split_host(parts.netloc)
+        if (
+            parts.scheme not in ("http", "https")
+            or origin_host.lower() not in allowed_hosts
+            or origin_port != port
+        ):
+            return Refusal(403, "origin_not_allowed", "request origin is not this application")
+
+    # Absent means a non-browser client; "none" is a user-initiated navigation.
+    fetch_site = headers.get("sec-fetch-site")
+    if fetch_site is not None and fetch_site not in ("same-origin", "none"):
+        return Refusal(403, "cross_site_blocked", "cross-site requests are not accepted")
+
+    length = headers.get("content-length")
+    if length and length.isdigit() and int(length) > max_request_bytes:
+        return Refusal(413, "payload_too_large", "request body exceeds the configured limit")
+
+    protected = path.startswith(API_PREFIX) and path not in PUBLIC_PATHS
+    if not protected:
+        return None
+
+    if headers.get("cookie-session") != session.id:
+        return Refusal(401, "unauthenticated", "a valid application session is required")
+    if method.upper() not in SAFE_METHODS and headers.get(CSRF_HEADER) != session.csrf_token:
+        return Refusal(403, "csrf_failed", "missing or invalid CSRF token")
+    return None
+
+
+class SecurityMiddleware:
+    """Pure-ASGI so the SSE stream keeps streaming (BaseHTTPMiddleware buffers)."""
+
+    def __init__(self, app, *, session: Session, config) -> None:
+        self.app = app
+        self.session = session
+        self.max_request_bytes = config.max_request_bytes
+        self.allowed_hosts = frozenset(LOOPBACK_HOSTS | {str(config.host).lower()})
+
+    async def __call__(self, scope, receive, send) -> None:
+        if scope["type"] != "http":
+            await self.app(scope, receive, send)
+            return
+        headers = Headers(scope=scope)
+        # The cookie is read here and handed to the pure policy as one value, so the
+        # policy never has to parse a Cookie header.
+        lookup = dict(headers)
+        lookup["cookie-session"] = _cookie(headers.get("cookie", ""), SESSION_COOKIE)
+        refusal = evaluate(
+            method=scope.get("method", "GET"),
+            path=scope.get("path", "/"),
+            headers=lookup,
+            session=self.session,
+            allowed_hosts=self.allowed_hosts,
+            max_request_bytes=self.max_request_bytes,
+        )
+        if refusal is not None:
+            response = JSONResponse(
+                status_code=refusal.status,
+                content={"error": {"code": refusal.code, "message": refusal.message}},
+                headers=DEFAULT_HEADERS,
+            )
+            await response(scope, receive, send)
+            return
+
+        async def send_with_headers(message):
+            if message["type"] == "http.response.start":
+                out = MutableHeaders(scope=message)
+                for name, value in DEFAULT_HEADERS.items():
+                    out.setdefault(name, value)
+            await send(message)
+
+        await self.app(scope, receive, send_with_headers)
+
+
+def _cookie(header: str, name: str) -> str:
+    for part in header.split(";"):
+        key, _, value = part.strip().partition("=")
+        if key == name:
+            return value
+    return ""

photo_pipeline/config.py

@@ -30,6 +30,10 @@ class Config(BaseModel):
     log_level: str = "INFO"
     log_format: str = "json"  # "json" or "text"
 
+    # Largest request body the API accepts. Every endpoint takes small JSON commands;
+    # anything larger is a mistake or an attempt to exhaust memory (US07-02).
+    max_request_bytes: int = 1_048_576
+
     # Library boundary for path validation (os.pathsep-separated in the env var).
     library_roots: tuple[Path, ...] = ()
     thumbnail_cache_quota_bytes: int = 500_000_000

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)

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

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

photo_pipeline/jobs/domain_handlers.py

@@ -38,7 +38,8 @@ def _safety_score_item(asset_id: str, ctx: JobContext) -> None:
 def _analysis_item(asset_id: str, ctx: JobContext) -> None:
     from photo_pipeline.services.analysis import AnalysisService
 
-    AnalysisService(ctx.session_factory).run([asset_id])
+    roots = tuple(getattr(ctx.config, "library_roots", ()) or ())
+    AnalysisService(ctx.session_factory, library_roots=roots).run([asset_id])
 
 
 def _upload_batch_item(batch_id: str, ctx: JobContext) -> None:

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

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

photo_pipeline/path_policy.py

@@ -49,6 +49,29 @@ def resolve_within(root: Path, path: os.PathLike | str) -> Path:
     return resolved
 
 
+def resolve_in_roots(roots: Iterable[os.PathLike | str], path: os.PathLike | str) -> Path:
+    """The resolved path, proven to be inside one of ``roots`` and not excluded.
+
+    Callers must use the **returned** path for whatever they do next: validating one
+    name and then opening another is the symlink race this exists to close (US07-02).
+    The message names no path — it reaches API responses.
+
+    With no roots configured there is no boundary to check; that is a property of the
+    configuration, not permission granted to this call.
+    """
+    if is_excluded(path):
+        raise PathPolicyError("path is inside an excluded (_IGNORE/) tree")
+    roots = list(roots)
+    if not roots:
+        return Path(path)
+    for root in roots:
+        try:
+            return resolve_within(Path(root), path)
+        except PathPolicyError:
+            continue
+    raise PathPolicyError("path is outside the configured library roots")
+
+
 def iter_supported_files(root: os.PathLike | str) -> Iterator[Path]:
     """Yield supported, non-excluded files under ``root`` in deterministic order.
 

photo_pipeline/services/analysis.py

@@ -25,9 +25,9 @@ from typing import Protocol
 from sqlalchemy import func, select
 from sqlalchemy.orm import sessionmaker
 
-from photo_pipeline.integrations import exiftool
+from photo_pipeline import path_policy
 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"
@@ -59,9 +59,16 @@ def _now() -> datetime:
 
 
 class AnalysisService:
-    def __init__(self, session_factory: sessionmaker, *, provider: VisionProvider | None = None) -> None:
+    def __init__(
+        self,
+        session_factory: sessionmaker,
+        *,
+        provider: VisionProvider | None = None,
+        library_roots: tuple = (),
+    ) -> None:
         self._session_factory = session_factory
         self._provider = provider
+        self._roots = tuple(library_roots)
 
     def _sfw_asset_ids(self, session) -> set[str]:
         """Asset ids whose latest safety decision is ``sfw`` — the ONLY assets that
@@ -136,6 +143,18 @@ class AnalysisService:
             if not path:
                 skipped += 1
                 continue
+            # Second gate, at the moment of use: the database says where the file
+            # was, the filesystem decides what that name means now. A link swapped
+            # under an asset after the scan would otherwise send bytes from outside
+            # the library — the one place that leaves this machine (US07-02).
+            try:
+                path = str(path_policy.resolve_in_roots(self._roots, path))
+            except path_policy.PathPolicyError as error:
+                self._store(
+                    asset_id, status="error", result=None, error=str(error), tokens=0, raw=""
+                )
+                errors += 1
+                continue
             try:
                 result = provider.analyze(path, album_hint=_album_hint(path))
             except Exception as error:  # provider/validation failure is per-asset
@@ -150,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}
 
@@ -175,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)
@@ -191,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 []

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

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)

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:

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).
@@ -129,7 +133,10 @@ class ThumbnailService:
                 raise ThumbnailUnavailable(f"asset {asset_id} has no readable file")
             source = str(source)
             if source == asset.current_path:
-                self._validate_path(source)  # archive roots lie outside the library
+                # Render the *resolved* path the check approved: revalidating and then
+                # reopening the original name would let a symlink swapped in between
+                # the two steps decide which bytes are served (US07-02).
+                source = str(self._validate_path(source))  # archive roots lie outside
 
         # Rendering happens outside the DB session (no transaction held during I/O).
         try:
@@ -195,20 +202,16 @@ class ThumbnailService:
                 session.commit()
 
     # ── path safety ──────────────────────────────────────────────────────────
-    def _validate_path(self, current_path: str) -> None:
-        path = Path(current_path)
-        if path_policy.is_excluded(path):
-            raise PathNotAllowed(f"excluded path: {current_path}")
-        roots = self._config.library_roots
-        if not roots:
-            return
-        for root in roots:
+    def _validate_path(self, current_path: str) -> Path:
+        """The resolved path to read, or ``PathNotAllowed``.
+
+        The message names no path: a refusal is returned to the browser, and where
+        the library lives is not the caller's business (US07-02).
+        """
         try:
-                path_policy.resolve_within(Path(root), path)
-                return
-            except path_policy.PathPolicyError:
-                continue
-        raise PathNotAllowed(f"path outside configured roots: {current_path}")
+            return path_policy.resolve_in_roots(self._config.library_roots, current_path)
+        except path_policy.PathPolicyError as error:
+            raise PathNotAllowed(str(error)) from None
 
     # ── cache key + rendering ──────────────────────────────────────────────────
     @staticmethod
@@ -223,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:
-        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"
-                    )
-                oriented = ImageOps.exif_transpose(image)
-                mode = "RGBA" if _has_alpha(oriented) else "RGB"
-                converted = oriented.convert(mode)
-                converted.thumbnail((size, size), Image.LANCZOS)
-                out_width, out_height = converted.size
+        """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}.tmp")
+        tmp = destination.with_name(f".{destination.name}.{uuid.uuid4().hex}{TEMP_SUFFIX}")
+        try:
+            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 = imaging.to_srgb(oriented, mode=mode)
+                converted.thumbnail((size, size), Image.LANCZOS)
+                out_width, out_height = converted.size
                 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 {
@@ -254,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

tests/conftest.py

@@ -1,4 +1,15 @@
-"""Make the repository root importable for the pipeline test suites."""
+"""Make the repository root importable for the pipeline test suites, and give every
+suite the application session the API requires since US07-02.
+
+The suites drive the API the way the browser does — module-level ``httpx`` calls and
+``TestClient`` — so instead of threading a cookie through several hundred call sites,
+both clients bootstrap the session themselves exactly like ``frontend/js/api.js``:
+fetch ``/api/v1/session`` once, then send the cookie plus the CSRF header, and
+re-bootstrap once on 401 (a restarted server issues a new session).
+
+Security tests deliberately bypass this by constructing their own ``httpx.Client``;
+only the module-level helpers are wrapped.
+"""
 
 import sys
 from pathlib import Path
@@ -6,3 +17,85 @@ from pathlib import Path
 REPO = Path(__file__).resolve().parents[1]
 if str(REPO) not in sys.path:
     sys.path.insert(0, str(REPO))
+
+import httpx
+import httpx._api  # httpx.get/post resolve `request` in this module
+import pytest
+from starlette.testclient import TestClient
+
+from photo_pipeline.api.security import CSRF_HEADER, SESSION_COOKIE
+
+_SESSIONS: dict[str, tuple[str, str]] = {}  # base url -> (session id, csrf token)
+
+
+def _base(url) -> str:
+    parts = httpx.URL(str(url))
+    return f"{parts.scheme}://{parts.netloc.decode()}"
+
+
+def _bootstrap(base: str) -> tuple[str, str]:
+    if base not in _SESSIONS:
+        response = httpx.Client(timeout=10).get(f"{base}/api/v1/session")
+        _SESSIONS[base] = (
+            response.cookies.get(SESSION_COOKIE, ""),
+            response.json().get("csrf_token", ""),
+        )
+    return _SESSIONS[base]
+
+
+def session_client(base_url: str, **kwargs) -> httpx.Client:
+    """An ``httpx.Client`` that has bootstrapped its own application session.
+
+    For suites that keep one client for a whole journey; it holds its own cookie, so
+    a client made after a server restart picks up the new session by construction.
+    """
+    client = httpx.Client(base_url=base_url, **kwargs)
+    client.headers[CSRF_HEADER] = client.get("/api/v1/session").json()["csrf_token"]
+    return client
+
+
+def _authorized(url, headers, cookies):
+    session_id, token = _bootstrap(_base(url))
+    headers = dict(headers or {})
+    headers.setdefault(CSRF_HEADER, token)
+    cookies = dict(cookies or {})
+    cookies.setdefault(SESSION_COOKIE, session_id)
+    return headers, cookies
+
+
+@pytest.fixture(autouse=True, scope="session")
+def _api_session():
+    real_request, real_stream = httpx._api.request, httpx._api.stream
+    real_client_request, real_client_init = TestClient.request, TestClient.__init__
+
+    def request(method, url, *, headers=None, cookies=None, **kwargs):
+        sent, jar = _authorized(url, headers, cookies)
+        response = real_request(method, url, headers=sent, cookies=jar, **kwargs)
+        if response.status_code == 401:
+            _SESSIONS.pop(_base(url), None)
+            sent, jar = _authorized(url, headers, cookies)
+            response = real_request(method, url, headers=sent, cookies=jar, **kwargs)
+        return response
+
+    def stream(method, url, *, headers=None, cookies=None, **kwargs):
+        sent, jar = _authorized(url, headers, cookies)
+        return real_stream(method, url, headers=sent, cookies=jar, **kwargs)
+
+    def client_init(self, app, *args, base_url="http://127.0.0.1", **kwargs):
+        # The default "http://testserver" is not a local host, which is exactly what
+        # the Host check refuses; in-process tests are still same-origin callers.
+        real_client_init(self, app, *args, base_url=base_url, **kwargs)
+
+    def client_request(self, method, url, *, headers=None, **kwargs):
+        if CSRF_HEADER not in self.headers:
+            response = real_client_request(self, "GET", "/api/v1/session")
+            self.headers[CSRF_HEADER] = response.json()["csrf_token"]
+        return real_client_request(self, method, url, headers=headers, **kwargs)
+
+    httpx._api.request, httpx.request = request, request
+    httpx._api.stream, httpx.stream = stream, stream
+    TestClient.request, TestClient.__init__ = client_request, client_init
+    yield
+    httpx._api.request, httpx.request = real_request, real_request
+    httpx._api.stream, httpx.stream = real_stream, real_stream
+    TestClient.request, TestClient.__init__ = real_client_request, real_client_init

tests/e2e/test_phase_a_pipeline.py

@@ -22,6 +22,8 @@ import numpy as np
 import pytest
 from PIL import Image
 
+from tests.conftest import session_client
+
 REPO = Path(__file__).resolve().parents[2]
 
 
@@ -92,7 +94,7 @@ class ServerController:
                 pytest.fail(f"server exited: {err.decode(errors='replace')}")
             try:
                 if httpx.get(f"{self.base}/api/v1/health/ready", timeout=1).status_code == 200:
-                    self.client = httpx.Client(base_url=self.base, timeout=10)
+                    self.client = session_client(self.base, timeout=10)
                     return
             except httpx.HTTPError:
                 time.sleep(0.2)

tests/e2e/test_review_ui.py

@@ -17,6 +17,9 @@ import httpx
 import numpy as np
 import pytest
 from PIL import Image
+from playwright.sync_api import expect
+
+from tests.conftest import session_client
 
 REPO = Path(__file__).resolve().parents[2]
 
@@ -111,7 +114,7 @@ def server(tmp_path):
         proc.terminate()
         pytest.fail("server never became ready")
 
-    client = httpx.Client(base_url=base, timeout=5)
+    client = session_client(base, timeout=5)
 
     def cluster_by_method(method):
         clusters = client.get("/api/v1/duplicates/clusters").json()["items"]
@@ -148,9 +151,9 @@ def test_fuzzy_decision_requires_confirmation(page, server):
     page.get_by_test_id("confirm").wait_for()
     assert "open" in page.get_by_test_id("cluster-state").inner_text()
     page.get_by_test_id("confirm-yes").click()
-    page.wait_for_function(
-        "document.querySelector('[data-testid=cluster-state]').innerText.includes('dismissed')"
-    )
+    # A locator assertion, not wait_for_function: the app's CSP forbids eval, and
+    # Playwright's polling predicate is evaluated as a string in the page (US07-02).
+    expect(page.get_by_test_id("cluster-state")).to_contain_text("dismissed")
 
 
 def test_stale_version_shows_conflict(page, server):
@@ -176,9 +179,9 @@ def test_decision_persists_after_reload(page, server):
     page.goto(f"{server.base}/app/#/duplicates/{cluster['id']}")
     page.get_by_test_id("not-duplicate").click()
     page.get_by_test_id("confirm-yes").click()
-    page.wait_for_function(
-        "document.querySelector('[data-testid=cluster-state]').innerText.includes('dismissed')"
-    )
+    # A locator assertion, not wait_for_function: the app's CSP forbids eval, and
+    # Playwright's polling predicate is evaluated as a string in the page (US07-02).
+    expect(page.get_by_test_id("cluster-state")).to_contain_text("dismissed")
     page.reload()
     page.get_by_test_id("cluster-state").wait_for()
     assert "dismissed" in page.get_by_test_id("cluster-state").inner_text()

tests/e2e/test_security.py

@@ -0,0 +1,280 @@
+"""US07-02 black-box security tests: authorization and path boundaries.
+
+Everything here talks to a real ``photo_pipeline serve`` child process over HTTP with
+its own ``httpx.Client``, deliberately outside the session helper the rest of the
+suite uses — an attacker does not get a bootstrapped client.
+
+The threat is a page in the user's browser, not a remote attacker: the app listens on
+127.0.0.1, so any site the user visits can send requests to it and can point an
+``<img>`` at its media endpoints. Each journey below is one of those attempts.
+"""
+
+from __future__ import annotations
+
+import json
+from contextlib import contextmanager
+from pathlib import Path
+
+import httpx
+import pytest
+
+from tests.e2e._pipeline_harness import Server, image, seed_library
+
+TIMEOUT = 20
+SENTINEL_KEY = "immich-sentinel-9f3a2b"
+SESSION_COOKIE = "pp_session"
+CSRF_HEADER = "X-CSRF-Token"
+
+# A GET, a mutation, and a media endpoint: the three shapes the policy must cover.
+PROTECTED = [
+    ("GET", "/api/v1/workflow", None),
+    ("POST", "/api/v1/albums/proposals", {}),
+    ("GET", "/api/v1/assets/{asset}/thumbnail?size=256", None),
+]
+
+
+@pytest.fixture(scope="module")
+def stack(tmp_path_factory):
+    """One server, one album of two photos, and a credential sentinel in its config."""
+    tmp_path = tmp_path_factory.mktemp("security")
+    seeded = seed_library(tmp_path, {"a": 1, "b": 2}, {"a": "sfw", "b": "sfw"})
+    outside = tmp_path / "outside"
+    outside.mkdir()
+    image(outside / "secret.jpg", 99)
+    server = Server(
+        seeded,
+        extra_env={
+            "PHOTO_PIPELINE_IMMICH_API_KEY": SENTINEL_KEY,
+            "PHOTO_PIPELINE_IMMICH_SERVER_URL": "http://127.0.0.1:1",
+        },
+    ).start()
+    yield server, seeded, outside
+    server.stop()
+
+
+@contextmanager
+def anonymous(server):
+    with httpx.Client(base_url=server.base, timeout=TIMEOUT) as client:
+        yield client
+
+
+@contextmanager
+def authenticated(server):
+    """A browser that has loaded the app: session cookie in the jar, token in a header."""
+    with httpx.Client(base_url=server.base, timeout=TIMEOUT) as client:
+        client.headers[CSRF_HEADER] = client.get("/api/v1/session").json()["csrf_token"]
+        yield client
+
+
+def call(client, method, path, body, asset):
+    return client.request(method, path.format(asset=asset), json=body)
+
+
+def test_the_api_refuses_every_caller_without_a_session(stack):
+    server, seeded, _ = stack
+    with anonymous(server) as client:
+        for method, path, body in PROTECTED:
+            response = call(client, method, path, body, seeded.asset_ids["a"])
+            assert response.status_code == 401, path
+            assert response.json()["error"]["code"] == "unauthenticated"
+        # Liveness and readiness stay open: an orchestrator holds no session.
+        assert client.get("/api/v1/health/ready").status_code == 200
+        assert client.get("/api/v1/health/live").status_code == 200
+
+
+def test_a_guessed_session_cookie_is_refused(stack):
+    server, _, _ = stack
+    with anonymous(server) as client:
+        client.cookies.set(SESSION_COOKIE, "guessed", domain="127.0.0.1")
+        response = client.get("/api/v1/workflow")
+        assert response.status_code == 401
+        assert response.json()["error"]["code"] == "unauthenticated"
+
+
+def test_the_bootstrap_issues_a_strict_httponly_cookie(stack):
+    server, _, _ = stack
+    with anonymous(server) as client:
+        response = client.get("/api/v1/session")
+        cookie = response.headers["set-cookie"].lower()
+        assert "httponly" in cookie and "samesite=strict" in cookie and "path=/" in cookie
+        assert response.json()["csrf_token"]
+        # The token is in the body, which no other origin may read: no CORS header
+        # grants access to it.
+        assert "access-control-allow-origin" not in response.headers
+
+
+def test_a_session_without_the_csrf_token_may_read_but_not_mutate(stack):
+    server, _, _ = stack
+    with authenticated(server) as client:
+        del client.headers[CSRF_HEADER]
+        assert client.get("/api/v1/albums/proposals").status_code == 200
+
+        response = client.post("/api/v1/albums/proposals", json={})
+        assert response.status_code == 403
+        assert response.json()["error"]["code"] == "csrf_failed"
+
+        response = client.post(
+            "/api/v1/albums/proposals", json={}, headers={CSRF_HEADER: "guessed"}
+        )
+        assert response.status_code == 403
+    # And nothing was created behind the refusal.
+    with authenticated(server) as client:
+        assert client.get("/api/v1/albums/proposals").json()["items"] == []
+
+
+def test_a_foreign_origin_cannot_mutate_even_with_a_session(stack):
+    server, _, _ = stack
+    with authenticated(server) as client:
+        for origin in ("http://evil.example", "http://127.0.0.1:1", "null"):
+            response = client.post(
+                "/api/v1/albums/proposals", json={}, headers={"Origin": origin}
+            )
+            assert response.status_code == 403, origin
+            assert response.json()["error"]["code"] == "origin_not_allowed"
+        # This app's own origin is accepted, so the check is not simply refusing all.
+        allowed = client.post(
+            "/api/v1/albums/proposals", json={}, headers={"Origin": server.base}
+        )
+        assert allowed.status_code == 200
+
+
+def test_a_rebinding_host_is_refused(stack):
+    """A name that resolves to 127.0.0.1 makes the browser treat the attacker's page
+    as same-origin. The Host header still carries that name, so it is checked."""
+    server, _, _ = stack
+    with authenticated(server) as client:
+        response = client.get("/api/v1/workflow", headers={"Host": "photos.evil.example"})
+        assert response.status_code == 403
+        assert response.json()["error"]["code"] == "host_not_allowed"
+
+
+def test_media_cannot_be_embedded_by_another_page(stack):
+    server, seeded, _ = stack
+    url = f"/api/v1/assets/{seeded.asset_ids['a']}/thumbnail?size=256"
+    with authenticated(server) as client:
+        # What an <img> on another site produces: no Origin, but a cross-site marker.
+        blocked = client.get(url, headers={"Sec-Fetch-Site": "cross-site"})
+        assert blocked.status_code == 403
+        assert blocked.json()["error"]["code"] == "cross_site_blocked"
+
+        served = client.get(url, headers={"Sec-Fetch-Site": "same-origin"})
+        assert served.status_code == 200
+        assert served.headers["content-type"] == "image/webp"
+        assert served.headers["cross-origin-resource-policy"] == "same-origin"
+        # Photos must never land in a shared cache.
+        assert served.headers["cache-control"].startswith("private")
+
+
+def test_every_response_carries_the_default_headers_and_no_cors(stack):
+    server, seeded, _ = stack
+    with authenticated(server) as client:
+        responses = [
+            client.get("/api/v1/workflow"),
+            client.get(f"/api/v1/assets/{seeded.asset_ids['a']}/thumbnail?size=256"),
+            client.get("/app/"),
+            client.get("/api/v1/does-not-exist"),
+        ]
+    for response in responses:
+        assert response.headers["x-content-type-options"] == "nosniff"
+        assert response.headers["x-frame-options"] == "DENY"
+        assert response.headers["referrer-policy"] == "no-referrer"
+        assert "frame-ancestors 'none'" in response.headers["content-security-policy"]
+        assert "access-control-allow-origin" not in response.headers
+        assert "access-control-allow-credentials" not in response.headers
+
+
+def test_a_traversal_attempt_addresses_nothing(stack):
+    """Identifiers are database keys, not paths: traversal has nowhere to land."""
+    server, _, _ = stack
+    with authenticated(server) as client:
+        for path in (
+            "/api/v1/albums/proposals/..%2F..%2F..%2Fetc%2Fpasswd",
+            "/api/v1/albums/proposals/../../../etc/passwd",
+            "/api/v1/assets/..%2F..%2Fetc%2Fpasswd/thumbnail?size=256",
+            "/api/v1/assets/%2Fetc%2Fpasswd/thumbnail?size=256",
+        ):
+            response = client.get(path)
+            assert response.status_code in (404, 422), path
+            assert "root:" not in response.text, path
+
+
+def test_an_oversized_request_is_refused_before_it_is_parsed(stack):
+    server, _, _ = stack
+    with authenticated(server) as client:
+        response = client.post(
+            "/api/v1/albums/proposals",
+            content=json.dumps({"albums": ["x" * 2_000_000]}),
+            headers={"Content-Type": "application/json"},
+        )
+        assert response.status_code == 413
+        assert response.json()["error"]["code"] == "payload_too_large"
+
+
+def test_a_malformed_request_reports_the_field_and_nothing_else(stack):
+    server, _, _ = stack
+    with authenticated(server) as client:
+        for content in ("{", '{"albums": 5}', ""):
+            response = client.post(
+                "/api/v1/albums/proposals",
+                content=content,
+                headers={"Content-Type": "application/json"},
+            )
+            assert response.status_code == 422, content
+            error = response.json()["error"]
+            assert error["code"] == "invalid_request"
+            assert "Traceback" not in response.text and "photo_pipeline/" not in response.text
+
+
+def test_errors_reveal_neither_the_credential_nor_an_internal(stack):
+    """The API key is configured but must appear nowhere; failures additionally say
+    nothing about where the library lives or how the server is built.
+
+    Successful responses are a different matter: path previews are the point of the
+    rename and archive views, and the operator is the one who owns those paths.
+    """
+    server, seeded, _ = stack
+    with authenticated(server) as client:
+        responses = [
+            client.post("/api/v1/upload-preflight", json={}),
+            client.get("/api/v1/workflow"),
+            client.get("/api/v1/upload-batches"),
+            client.get("/api/v1/assets/unknown-asset/thumbnail?size=256"),
+            client.get("/api/v1/archive-plans/unknown-plan"),
+            client.post("/api/v1/archive-locations", json={"name": "x", "root": "/nope"}),
+            client.get("/api/v1/does-not-exist"),
+        ]
+    for response in responses:
+        assert SENTINEL_KEY not in response.text
+        assert "Traceback" not in response.text
+        if response.status_code >= 400:
+            assert str(seeded.lib) not in response.text
+            assert "photo_pipeline/" not in response.text
+            assert "sqlite" not in response.text.lower()
+
+
+def test_a_symlink_swapped_under_an_asset_cannot_be_served(stack):
+    """TOCTOU on the media path: the file the database points at is replaced by a
+    link to something outside the library between the scan and the request."""
+    server, seeded, outside = stack
+    secret = outside / "secret.jpg"
+    original = seeded.lib / "b.jpg"
+    original.unlink()
+    original.symlink_to(secret)
+
+    with authenticated(server) as client:
+        response = client.get(f"/api/v1/assets/{seeded.asset_ids['b']}/thumbnail?size=256")
+    assert response.status_code == 403
+    assert response.json()["error"]["code"] == "path_not_allowed"
+    # The refusal names no filesystem location, and no bytes escaped with it.
+    assert str(outside) not in response.text and str(seeded.lib) not in response.text
+    assert secret.read_bytes()[:16] not in response.content
+
+
+def test_the_frontend_shell_stays_reachable_without_a_session(stack):
+    """It must load before any JavaScript can ask for a session."""
+    server, _, _ = stack
+    with anonymous(server) as client:
+        response = client.get("/app/", headers={"Sec-Fetch-Site": "none"})
+        assert response.status_code == 200
+        assert response.headers["content-type"].startswith("text/html")
+        assert Path("frontend/index.html").exists()

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

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

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)

tests/integration/test_security_boundaries.py

@@ -0,0 +1,136 @@
+"""US07-02: what an unhandled failure says, and where the library ends.
+
+Two boundaries that only show up below the HTTP surface:
+
+* Exception text is where internals leak — absolute paths, SQL, and occasionally a
+  credential passed to the call that blew up. The first tests drive the real
+  application with a route that raises such an exception (no production route does)
+  and assert the client sees only a code.
+* The path the database recorded is not the path the filesystem will open a moment
+  later. Analysis is the one stage whose bytes leave this machine, so it resolves
+  the source against the library roots immediately before the provider call.
+"""
+
+from __future__ import annotations
+
+import logging
+import uuid
+from datetime import datetime, timezone
+
+import pytest
+from PIL import Image
+from starlette.testclient import TestClient
+
+from photo_pipeline.api.app import create_app
+from photo_pipeline.config import Config
+from photo_pipeline.db import create_db_engine, create_session_factory, run_migrations
+from photo_pipeline.models import AnalysisResult, Asset
+from photo_pipeline.services.analysis import AnalysisService
+from photo_pipeline.services.safety import SafetyService
+
+BOOM = "sqlite:///Users/someone/Pictures/private.db failed with key sk-secret-123"
+
+
+@pytest.fixture
+def client(tmp_path):
+    app = create_app(Config(data_dir=tmp_path / "data"))
+
+    @app.get("/api/v1/boom")
+    def boom():
+        raise RuntimeError(BOOM)
+
+    with TestClient(app, raise_server_exceptions=False) as test_client:
+        yield test_client
+
+
+def test_an_unhandled_error_returns_a_bare_envelope(client, caplog):
+    with caplog.at_level(logging.ERROR):
+        response = client.get("/api/v1/boom")
+
+    assert response.status_code == 500
+    assert response.json() == {"error": {"code": "internal_error", "message": "internal error"}}
+    assert BOOM not in response.text and "Traceback" not in response.text
+    # The operator still gets the whole story, on the server side.
+    assert BOOM in caplog.text
+
+
+def test_a_refusal_response_still_carries_the_default_headers(client):
+    """A 500 escaping the middleware's response path would also escape its headers."""
+    response = client.get("/api/v1/boom")
+    assert response.headers["x-content-type-options"] == "nosniff"
+    assert response.headers["x-frame-options"] == "DENY"
+
+
+# ── the library boundary, revalidated at the moment of use ───────────────────
+
+
+class RecordingProvider:
+    def __init__(self):
+        self.calls = []
+
+    def analyze(self, path, *, album_hint):
+        self.calls.append(path)
+        return {"description": "a photo", "tags": []}
+
+
+def _library(tmp_path):
+    lib = tmp_path / "lib"
+    lib.mkdir()
+    outside = tmp_path / "outside"
+    outside.mkdir()
+    for path in (lib / "inside.jpg", outside / "private.jpg"):
+        Image.new("RGB", (8, 8), "blue").save(path)
+    (tmp_path / "data").mkdir()
+    config = Config.from_env({"PHOTO_PIPELINE_DATA_DIR": str(tmp_path / "data")})
+    run_migrations(config.database_url)
+    return lib, outside, create_session_factory(create_db_engine(config.database_url))
+
+
+def _sfw_asset(sf, path):
+    asset = Asset(
+        id=str(uuid.uuid4()),
+        original_path=str(path),
+        current_path=str(path),
+        discovered_at=datetime.now(timezone.utc),
+        hash_version=1,
+    )
+    with sf() as session:
+        session.add(asset)
+        session.commit()
+    SafetyService(sf).decide(asset.id, "sfw", write_exif=False)
+    return asset.id
+
+
+def test_analysis_will_not_send_a_file_that_left_the_library(tmp_path):
+    """A link swapped under an asset after the scan points at something the user
+    never put in the library. Those bytes must not reach the vision provider — it is
+    the one place in the pipeline where content leaves this machine."""
+    lib, outside, sf = _library(tmp_path)
+    inside = lib / "inside.jpg"
+    asset_id = _sfw_asset(sf, inside)
+    inside.unlink()
+    inside.symlink_to(outside / "private.jpg")
+
+    provider = RecordingProvider()
+    result = AnalysisService(sf, provider=provider, library_roots=(lib,)).run([asset_id])
+
+    assert provider.calls == [], "the provider must never have been constructed a request"
+    assert result == {"analyzed": 0, "skipped": 0, "errors": 1}
+    with sf() as session:
+        row = session.get(AnalysisResult, asset_id)
+    # The failure is visible and names no path.
+    assert row.status == "error"
+    assert "outside the configured library roots" in row.error_message
+    assert str(outside) not in row.error_message
+
+
+def test_analysis_still_reads_a_file_that_stayed_inside(tmp_path):
+    """The guard must resolve real paths, not refuse everything."""
+    lib, _, sf = _library(tmp_path)
+    asset_id = _sfw_asset(sf, lib / "inside.jpg")
+
+    provider = RecordingProvider()
+    result = AnalysisService(sf, provider=provider, library_roots=(lib,)).run([asset_id])
+
+    assert result["analyzed"] == 1
+    assert provider.calls == [str((lib / "inside.jpg").resolve())]

tests/story_traceability.json

@@ -147,6 +147,15 @@
     "US07-01": [
       "tests/unit/test_legacy_archive.py",
       "tests/integration/test_legacy_import.py"
+    ],
+    "US07-02": [
+      "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"
     ]
   }
 }

tests/unit/test_path_policy.py

@@ -81,3 +81,42 @@ def test_symlink_within_root_is_allowed(tmp_path):
         pytest.skip("cannot create symlink on this platform")
     found = path_policy.discover([root])
     assert target in found and link in found
+
+
+def test_resolve_in_roots_returns_the_path_the_caller_must_use(tmp_path):
+    """The resolved path is the answer, not a yes/no: a caller that revalidates one
+    name and then opens another has an open symlink race (US07-02)."""
+    root = tmp_path / "lib"
+    (root / "sub").mkdir(parents=True)
+    target = root / "sub" / "real.jpg"
+    target.write_bytes(b"x")
+    link = root / "alias.jpg"
+    os.symlink(target, link)
+
+    assert path_policy.resolve_in_roots([root], link) == target.resolve()
+    assert path_policy.resolve_in_roots([root], target) == target.resolve()
+
+
+def test_resolve_in_roots_refuses_escapes_without_naming_them(tmp_path):
+    root = tmp_path / "lib"
+    root.mkdir()
+    outside = tmp_path / "outside.jpg"
+    outside.write_bytes(b"x")
+    link = root / "alias.jpg"
+    os.symlink(outside, link)
+
+    for candidate in (link, outside, root / "_IGNORE" / "a.jpg"):
+        with pytest.raises(path_policy.PathPolicyError) as raised:
+            path_policy.resolve_in_roots([root], candidate)
+        assert str(outside) not in str(raised.value)
+
+
+def test_resolve_in_roots_checks_every_configured_root(tmp_path):
+    first, second = tmp_path / "one", tmp_path / "two"
+    first.mkdir()
+    second.mkdir()
+    photo = second / "b.jpg"
+    photo.write_bytes(b"x")
+    assert path_policy.resolve_in_roots([first, second], photo) == photo.resolve()
+    # No configured boundary means nothing to check against.
+    assert path_policy.resolve_in_roots([], photo) == photo

tests/unit/test_security_policy.py

@@ -0,0 +1,169 @@
+"""US07-02: the request-admission policy, enumerated.
+
+``evaluate`` decides every refusal the API can make before a route runs, so the
+whole local-web threat model is one table here: who may call, from where, with what
+proof. The middleware and the endpoints are covered black box in
+``tests/e2e/test_security.py``; this file pins the rules themselves, including the
+combinations a browser can produce but a test client rarely does.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from photo_pipeline.api.security import (
+    CSRF_HEADER,
+    PUBLIC_PATHS,
+    Session,
+    evaluate,
+    split_host,
+)
+
+SESSION = Session(id="session-id", csrf_token="csrf-token")
+HOST = "127.0.0.1:8000"
+LIMIT = 1024
+
+
+def check(method="GET", path="/api/v1/workflow", **headers):
+    """Evaluate a request that is authenticated and same-origin unless overridden."""
+    sent = {
+        "host": HOST,
+        "cookie-session": SESSION.id,
+        CSRF_HEADER: SESSION.csrf_token,
+    }
+    sent.update({name.replace("_", "-"): value for name, value in headers.items()})
+    sent = {name: value for name, value in sent.items() if value is not None}
+    return evaluate(
+        method=method,
+        path=path,
+        headers=sent,
+        session=SESSION,
+        max_request_bytes=LIMIT,
+    )
+
+
+def test_an_authenticated_same_origin_request_is_admitted():
+    assert check() is None
+    assert check(method="POST", origin="http://127.0.0.1:8000") is None
+    assert check(sec_fetch_site="same-origin") is None
+
+
+@pytest.mark.parametrize("host", ["evil.example", "evil.example:8000", "192.168.1.10:8000", ""])
+def test_a_non_loopback_host_is_refused(host):
+    """DNS rebinding: the browser thinks it is talking to the attacker's name, which
+    resolves to 127.0.0.1. The name is the evidence, so the name is checked."""
+    refusal = check(host=host)
+    assert (refusal.status, refusal.code) == (403, "host_not_allowed")
+
+
+@pytest.mark.parametrize("host", ["127.0.0.1:8000", "localhost:8000", "[::1]:8000", "localhost"])
+def test_loopback_hosts_are_accepted(host):
+    assert check(host=host) is None
+
+
+@pytest.mark.parametrize(
+    "origin",
+    [
+        "http://evil.example",
+        "https://evil.example:8000",
+        "http://127.0.0.1:9999",  # another local app is still another origin
+        "http://localhost.evil.example:8000",
+        "null",
+        "file://",
+    ],
+)
+def test_a_foreign_origin_is_refused(origin):
+    refusal = check(method="POST", origin=origin)
+    assert (refusal.status, refusal.code) == (403, "origin_not_allowed")
+
+
+@pytest.mark.parametrize("origin", ["http://127.0.0.1:8000", "http://localhost:8000"])
+def test_this_applications_origin_is_accepted(origin):
+    assert check(method="POST", origin=origin) is None
+
+
+@pytest.mark.parametrize("site", ["cross-site", "same-site"])
+def test_a_cross_site_fetch_is_refused_even_without_an_origin(site):
+    """What ``<img src="http://127.0.0.1:8000/...">`` on another page looks like."""
+    refusal = check(path="/api/v1/assets/a1/thumbnail", sec_fetch_site=site)
+    assert (refusal.status, refusal.code) == (403, "cross_site_blocked")
+
+
+def test_a_user_initiated_navigation_is_accepted():
+    assert check(sec_fetch_site="none") is None
+
+
+def test_a_request_without_a_session_is_unauthenticated():
+    for method, path in [
+        ("GET", "/api/v1/workflow"),
+        ("POST", "/api/v1/jobs"),
+        ("GET", "/api/v1/assets/a1/thumbnail"),
+    ]:
+        refusal = check(method=method, path=path, cookie_session=None)
+        assert (refusal.status, refusal.code) == (401, "unauthenticated"), path
+
+
+def test_a_forged_session_is_unauthenticated():
+    refusal = check(cookie_session="guessed")
+    assert (refusal.status, refusal.code) == (401, "unauthenticated")
+
+
+@pytest.mark.parametrize("path", sorted(PUBLIC_PATHS))
+def test_health_and_the_bootstrap_stay_reachable_without_a_session(path):
+    assert check(path=path, cookie_session=None) is None
+
+
+def test_the_static_shell_needs_no_session():
+    """It has to load before any JavaScript can ask for one."""
+    assert check(path="/app/index.html", cookie_session=None) is None
+
+
+@pytest.mark.parametrize("method", ["POST", "PUT", "PATCH", "DELETE"])
+def test_a_mutation_without_a_valid_csrf_token_is_refused(method):
+    for token in (None, "guessed"):
+        refusal = check(method=method, path="/api/v1/jobs", **{CSRF_HEADER: token})
+        assert (refusal.status, refusal.code) == (403, "csrf_failed")
+
+
+@pytest.mark.parametrize("method", ["GET", "HEAD", "OPTIONS"])
+def test_reads_need_no_csrf_token(method):
+    assert check(method=method, **{CSRF_HEADER: None}) is None
+
+
+def test_an_oversized_body_is_refused_before_it_is_read():
+    refusal = check(method="POST", path="/api/v1/jobs", content_length=str(LIMIT + 1))
+    assert (refusal.status, refusal.code) == (413, "payload_too_large")
+    assert check(method="POST", path="/api/v1/jobs", content_length=str(LIMIT)) is None
+
+
+def test_the_host_check_precedes_authentication():
+    """A refusal must not tell a foreign caller whether its session guess was right."""
+    refusal = check(host="evil.example", cookie_session="guessed")
+    assert refusal.code == "host_not_allowed"
+
+
+def test_refusals_name_no_path_secret_or_internal():
+    refusals = [
+        check(host="evil.example"),
+        check(method="POST", origin="http://evil.example"),
+        check(cookie_session=None),
+        check(method="POST", **{CSRF_HEADER: None}),
+    ]
+    for refusal in refusals:
+        assert SESSION.id not in refusal.message
+        assert SESSION.csrf_token not in refusal.message
+        assert "/" not in refusal.message
+
+
+@pytest.mark.parametrize(
+    "value,expected",
+    [
+        ("127.0.0.1:8000", ("127.0.0.1", "8000")),
+        ("localhost", ("localhost", "")),
+        ("[::1]:8000", ("[::1]", "8000")),
+        ("[::1]", ("[::1]", "")),
+        ("", ("", "")),
+    ],
+)
+def test_split_host(value, expected):
+    assert split_host(value) == expected