Compare commits

...

2 Commits

34 changed files with 2659 additions and 87 deletions

View File

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

View File

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

View File

@@ -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 });

View File

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

View File

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

View File

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

View File

@@ -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:

View File

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

View File

@@ -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"},
)

View File

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

View File

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

119
photo_pipeline/imaging.py Normal file
View File

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

View File

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

View File

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

View File

@@ -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:

View File

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

View File

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

View File

@@ -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.

View File

@@ -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 []

View File

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

View File

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

View File

@@ -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:

View File

@@ -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:
try:
path_policy.resolve_within(Path(root), path)
return
except path_policy.PathPolicyError:
continue
raise PathNotAllowed(f"path outside configured roots: {current_path}")
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:
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:
"""Decode bounded, orient, colour-manage, resize, and write atomically.
The temporary file is removed on every failure path: a decoder that dies
halfway through ``save`` would otherwise leave a stray ``.tmp`` in the cache
forever (US07-03).
"""
destination = self._cache_path(cache_key)
destination.parent.mkdir(parents=True, exist_ok=True)
tmp = destination.with_name(f".{destination.name}.{uuid.uuid4().hex}{TEMP_SUFFIX}")
try:
with Image.open(source) as image:
width, height = image.size
if width * height > self._config.thumbnail_max_pixels:
raise ImageTooLarge(
f"{width}x{height} exceeds {self._config.thumbnail_max_pixels} px"
)
with imaging.open_image(source, max_pixels=self._config.thumbnail_max_pixels) as image:
imaging.draft(image, size) # JPEG decodes near the target size
oriented = ImageOps.exif_transpose(image)
mode = "RGBA" if _has_alpha(oriented) else "RGB"
converted = oriented.convert(mode)
converted = imaging.to_srgb(oriented, mode=mode)
converted.thumbnail((size, size), Image.LANCZOS)
out_width, out_height = converted.size
destination = self._cache_path(cache_key)
destination.parent.mkdir(parents=True, exist_ok=True)
tmp = destination.with_name(f".{destination.name}.{uuid.uuid4().hex}.tmp")
converted.save(tmp, format="WEBP", quality=82, method=4)
except ImageTooLarge:
except imaging.ImageTooLarge as error:
tmp.unlink(missing_ok=True)
raise ImageTooLarge(str(error)) from None
except imaging.UndecodableImage as error:
tmp.unlink(missing_ok=True)
raise UnsupportedImage(str(error)) from None
except Exception:
tmp.unlink(missing_ok=True)
raise
except (UnidentifiedImageError, OSError, ValueError) as error:
raise UnsupportedImage(f"cannot decode {source}: {error}") from error
import os
os.replace(tmp, destination)
return {
@@ -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

View File

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

View File

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

View File

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

280
tests/e2e/test_security.py Normal file
View File

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

325
tests/fixtures/media_corpus.py vendored Normal file
View File

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

View File

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

View File

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

View File

@@ -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())]

View File

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

View File

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

View File

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