Phase 2: Bild-Upload, WebP-Konvertierung und Stapel-Import
Uploads werden geprüft, nach WebP gewandelt (max. 1200 px Breite) und über eine eigene Route ausgeliefert. Dazu der Stapel-Import: viele Fotos auf einmal hochladen, pro Foto entsteht ein Entwurf. Der Upload ist die grösste Angriffsfläche der Anwendung - die Galerie ist frei zugänglich, und hier landet fremdes Binärmaterial auf der Platte. Entsprechend: - Der Dateiname wird selbst erzeugt (UUID + .webp), nie aus dem Upload übernommen. Beim Ausliefern wird er zusätzlich gegen ein Muster geprüft: er kommt dort aus der URL und ist damit Nutzereingabe, auch wenn wir ihn ursprünglich vergeben haben. - Der Inhalt entscheidet, nicht die Endung - Pillow muss das Bild öffnen können. Eine als .jpg getarnte PHP-Datei fällt durch. - Pixelmasse ausdrücklich begrenzt, VOR dem Dekodieren. Pillow allein wirft erst bei der doppelten Zahl einen Fehler; dazwischen gibt es nur eine Warnung, die niemand sieht. - Gespeichert wird nur das Ergebnis. Das entfernt eingebettete Fremdinhalte und die EXIF-Daten - bei Handyfotos oft samt GPS-Koordinaten der Wohnung. Die Drehung wird vorher auf die Bildpunkte angewendet, sonst lägen Hochkant-Fotos quer. - Bilder liegen nicht unter static/, sondern werden mit festem Content-Type und nosniff ausgeliefert. HEIC wird unterstützt (pillow-heif): iPhones nehmen so auf, und ohne das fiele beim Stapel-Import jedes iPhone-Foto durch. Ein unlesbares Bild lässt bewusst den ganzen Stapel scheitern - bei zwanzig Fotos fiele sonst niemandem auf, dass eines fehlt. Beim Löschen werden die Dateien mitentfernt (ON DELETE CASCADE räumt nur die Zeilen weg), und wird das Vorschaubild gelöscht, rückt das nächste nach. 42 Tests, alle grün. Zusätzlich von Hand gegen die laufende API geprüft: 3000x2200-Foto kommt als WebP 1200x880 an, Pfadmanipulation und getarnte Dateien werden abgewiesen, Entwürfe bleiben aus der Galerie draussen. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01GR4bNaj9GtRu57J4Niii8o
This commit is contained in:
+34
@@ -192,3 +192,37 @@ def bild_anhaengen(db: Session, item: Item, dateiname: str) -> ItemBild:
|
|||||||
db.commit()
|
db.commit()
|
||||||
db.refresh(bild)
|
db.refresh(bild)
|
||||||
return bild
|
return bild
|
||||||
|
|
||||||
|
|
||||||
|
def hauptbild_setzen(db: Session, bild: ItemBild) -> ItemBild:
|
||||||
|
"""Macht dieses Bild zum Vorschaubild und nimmt das Kennzeichen den
|
||||||
|
übrigen Bildern desselben Eintrags weg."""
|
||||||
|
for anderes in db.scalars(
|
||||||
|
select(ItemBild).where(ItemBild.item_id == bild.item_id)
|
||||||
|
):
|
||||||
|
anderes.is_primary = anderes.id == bild.id
|
||||||
|
db.commit()
|
||||||
|
db.refresh(bild)
|
||||||
|
return bild
|
||||||
|
|
||||||
|
|
||||||
|
def bild_loeschen(db: Session, bild: ItemBild) -> str:
|
||||||
|
"""Löscht die Zeile und meldet den Dateinamen zurück, damit der
|
||||||
|
Aufrufer die Datei entfernen kann."""
|
||||||
|
dateiname = bild.image_url
|
||||||
|
war_hauptbild = bild.is_primary
|
||||||
|
item_id = bild.item_id
|
||||||
|
db.delete(bild)
|
||||||
|
db.commit()
|
||||||
|
|
||||||
|
# War es das Vorschaubild, rückt das nächste nach - sonst stünde der
|
||||||
|
# Eintrag in der Galerie ohne Bild da, obwohl noch welche vorhanden sind.
|
||||||
|
if war_hauptbild:
|
||||||
|
naechstes = db.scalars(
|
||||||
|
select(ItemBild).where(ItemBild.item_id == item_id).order_by(ItemBild.id)
|
||||||
|
).first()
|
||||||
|
if naechstes is not None:
|
||||||
|
naechstes.is_primary = True
|
||||||
|
db.commit()
|
||||||
|
|
||||||
|
return dateiname
|
||||||
|
|||||||
+128
@@ -0,0 +1,128 @@
|
|||||||
|
"""Bild-Uploads prüfen und nach WebP wandeln.
|
||||||
|
|
||||||
|
Der Upload ist die grösste Angriffsfläche der Anwendung - die Galerie ist
|
||||||
|
frei zugänglich, und hier landet fremdes Binärmaterial auf der Platte.
|
||||||
|
Vier Punkte sind dafür entscheidend:
|
||||||
|
|
||||||
|
1. **Der Dateiname wird selbst erzeugt**, nie aus dem Upload übernommen.
|
||||||
|
Damit ist Pfad-Manipulation ("../../etc/...") von vornherein ausgeschlossen.
|
||||||
|
2. **Der Inhalt entscheidet, nicht die Endung.** Pillow muss das Bild
|
||||||
|
tatsächlich öffnen können; eine als .jpg getarnte PHP-Datei fällt dabei
|
||||||
|
durch. Der vom Browser gemeldete Content-Type wird nicht geglaubt.
|
||||||
|
3. **Pixelmasse begrenzen.** Ein wenige Kilobyte grosses PNG kann beim
|
||||||
|
Entpacken Gigabyte belegen ("Dekompressionsbombe"). MAX_IMAGE_PIXELS
|
||||||
|
deckelt das, bevor der Speicher volläuft.
|
||||||
|
4. **Nur das Ergebnis wird gespeichert**, nie das Original. Das entfernt
|
||||||
|
eingebettete Fremdinhalte und die EXIF-Daten - letztere enthalten bei
|
||||||
|
Handyfotos oft die GPS-Koordinaten der Wohnung.
|
||||||
|
"""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import re
|
||||||
|
import uuid
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
|
from PIL import Image, ImageOps, UnidentifiedImageError
|
||||||
|
|
||||||
|
# HEIC-Unterstützung anmelden (iPhone-Aufnahmen).
|
||||||
|
try:
|
||||||
|
from pillow_heif import register_heif_opener
|
||||||
|
|
||||||
|
register_heif_opener()
|
||||||
|
except ImportError: # pragma: no cover - nur ohne optionales Paket
|
||||||
|
pass
|
||||||
|
|
||||||
|
# Deckel gegen Dekompressionsbomben: 50 Megapixel sind rund doppelt so viel
|
||||||
|
# wie eine aktuelle Handykamera liefert und damit unverdächtig.
|
||||||
|
MAX_PIXEL = 50_000_000
|
||||||
|
Image.MAX_IMAGE_PIXELS = MAX_PIXEL
|
||||||
|
|
||||||
|
MAX_BREITE = 1200
|
||||||
|
WEBP_QUALITAET = 82
|
||||||
|
|
||||||
|
# Muster der selbst erzeugten Namen. Wird beim Ausliefern geprüft: der
|
||||||
|
# Dateiname kommt dort aus der URL und ist damit Nutzereingabe, auch wenn
|
||||||
|
# wir ihn ursprünglich selbst vergeben haben.
|
||||||
|
DATEINAME_MUSTER = re.compile(r"^[0-9a-f]{32}\.webp$")
|
||||||
|
|
||||||
|
|
||||||
|
class BildFehler(ValueError):
|
||||||
|
"""Upload abgelehnt - die Meldung ist für Nutzende gedacht."""
|
||||||
|
|
||||||
|
|
||||||
|
def dateiname_gueltig(name: str) -> bool:
|
||||||
|
return bool(DATEINAME_MUSTER.match(name))
|
||||||
|
|
||||||
|
|
||||||
|
def _oeffnen(rohdaten: bytes) -> Image.Image:
|
||||||
|
from io import BytesIO
|
||||||
|
|
||||||
|
try:
|
||||||
|
bild = Image.open(BytesIO(rohdaten))
|
||||||
|
# verify() liest die Struktur, ohne die Pixel zu dekodieren - danach
|
||||||
|
# ist das Objekt unbrauchbar, deshalb anschliessend neu öffnen.
|
||||||
|
bild.verify()
|
||||||
|
bild = Image.open(BytesIO(rohdaten))
|
||||||
|
|
||||||
|
# Pillow wirft von sich aus erst bei der DOPPELTEN Pixelzahl einen
|
||||||
|
# Fehler; dazwischen gibt es nur eine Warnung, die niemand sieht.
|
||||||
|
# Darum hier ausdrücklich prüfen - und zwar VOR load(), denn erst
|
||||||
|
# das Dekodieren belegt den Speicher.
|
||||||
|
breite, hoehe = bild.size
|
||||||
|
if breite * hoehe > MAX_PIXEL:
|
||||||
|
raise BildFehler("Das Bild hat zu viele Bildpunkte.")
|
||||||
|
|
||||||
|
bild.load()
|
||||||
|
return bild
|
||||||
|
except BildFehler:
|
||||||
|
# Eigene Meldung durchlassen. Muss vor dem ValueError-Zweig stehen:
|
||||||
|
# BildFehler erbt davon und bekäme sonst die falsche Begründung.
|
||||||
|
raise
|
||||||
|
except Image.DecompressionBombError as e:
|
||||||
|
raise BildFehler("Das Bild hat zu viele Bildpunkte.") from e
|
||||||
|
except (UnidentifiedImageError, OSError, ValueError) as e:
|
||||||
|
raise BildFehler("Das ist keine lesbare Bilddatei.") from e
|
||||||
|
|
||||||
|
|
||||||
|
def speichern(rohdaten: bytes, ziel_ordner: Path, max_bytes: int) -> str:
|
||||||
|
"""Prüft, wandelt und legt ab. Gibt den erzeugten Dateinamen zurück."""
|
||||||
|
if not rohdaten:
|
||||||
|
raise BildFehler("Die Datei ist leer.")
|
||||||
|
if len(rohdaten) > max_bytes:
|
||||||
|
raise BildFehler(
|
||||||
|
f"Die Datei ist zu gross (höchstens {max_bytes // (1024 * 1024)} MB)."
|
||||||
|
)
|
||||||
|
|
||||||
|
bild = _oeffnen(rohdaten)
|
||||||
|
|
||||||
|
# Drehung aus den EXIF-Daten auf die Bildpunkte anwenden, bevor die
|
||||||
|
# Daten wegfallen. Ohne das lägen Hochkant-Fotos in der Galerie quer.
|
||||||
|
bild = ImageOps.exif_transpose(bild)
|
||||||
|
|
||||||
|
# WebP kann RGB und RGBA. Alles andere (Graustufen, Palette, CMYK)
|
||||||
|
# vorher umwandeln, sonst scheitert das Speichern.
|
||||||
|
if bild.mode not in ("RGB", "RGBA"):
|
||||||
|
bild = bild.convert("RGBA" if "A" in bild.getbands() else "RGB")
|
||||||
|
|
||||||
|
if bild.width > MAX_BREITE:
|
||||||
|
hoehe = round(bild.height * MAX_BREITE / bild.width)
|
||||||
|
bild = bild.resize((MAX_BREITE, hoehe), Image.LANCZOS)
|
||||||
|
|
||||||
|
ziel_ordner.mkdir(parents=True, exist_ok=True)
|
||||||
|
name = f"{uuid.uuid4().hex}.webp"
|
||||||
|
# Kein exif=, kein icc_profile: es wird ausschliesslich das Bild
|
||||||
|
# geschrieben, ohne die Zusatzdaten aus dem Original.
|
||||||
|
bild.save(ziel_ordner / name, "WEBP", quality=WEBP_QUALITAET, method=4)
|
||||||
|
return name
|
||||||
|
|
||||||
|
|
||||||
|
def loeschen(ziel_ordner: Path, name: str) -> None:
|
||||||
|
"""Entfernt eine Bilddatei; fehlt sie schon, ist das kein Fehler.
|
||||||
|
|
||||||
|
ON DELETE CASCADE räumt nur die Zeilen in item_images weg - ohne diesen
|
||||||
|
Aufruf liefe das Upload-Verzeichnis mit Waisen voll.
|
||||||
|
"""
|
||||||
|
if not dateiname_gueltig(name):
|
||||||
|
return
|
||||||
|
(ziel_ordner / name).unlink(missing_ok=True)
|
||||||
+2
-1
@@ -11,7 +11,7 @@ from fastapi import FastAPI
|
|||||||
|
|
||||||
from app.config import einstellungen
|
from app.config import einstellungen
|
||||||
from app.database import SessionLocal, engine
|
from app.database import SessionLocal, engine
|
||||||
from app.routers import categories, items
|
from app.routers import categories, images, items
|
||||||
from app.seed import stammdaten_anlegen
|
from app.seed import stammdaten_anlegen
|
||||||
|
|
||||||
|
|
||||||
@@ -37,6 +37,7 @@ app = FastAPI(
|
|||||||
|
|
||||||
app.include_router(categories.router)
|
app.include_router(categories.router)
|
||||||
app.include_router(items.router)
|
app.include_router(items.router)
|
||||||
|
app.include_router(images.router)
|
||||||
|
|
||||||
|
|
||||||
@app.get("/health", tags=["Betrieb"])
|
@app.get("/health", tags=["Betrieb"])
|
||||||
|
|||||||
@@ -0,0 +1,141 @@
|
|||||||
|
"""Bilder hochladen, ausliefern und löschen."""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
from fastapi import APIRouter, Depends, File, HTTPException, Response, UploadFile, status
|
||||||
|
from fastapi.responses import FileResponse
|
||||||
|
from sqlalchemy.orm import Session
|
||||||
|
|
||||||
|
from app import crud, images
|
||||||
|
from app.config import einstellungen
|
||||||
|
from app.database import get_db
|
||||||
|
from app.models import Item, ItemBild
|
||||||
|
from app.schemas import BildAus, ItemAus
|
||||||
|
|
||||||
|
router = APIRouter(tags=["Bilder"])
|
||||||
|
|
||||||
|
|
||||||
|
def _hochladen(datei: UploadFile) -> str:
|
||||||
|
cfg = einstellungen()
|
||||||
|
rohdaten = datei.file.read(cfg.max_upload_bytes + 1)
|
||||||
|
try:
|
||||||
|
return images.speichern(rohdaten, cfg.upload_pfad, cfg.max_upload_bytes)
|
||||||
|
except images.BildFehler as e:
|
||||||
|
raise HTTPException(status_code=422, detail=str(e)) from e
|
||||||
|
|
||||||
|
|
||||||
|
# ------------------------------------------------------------- Ausliefern ---
|
||||||
|
|
||||||
|
@router.get("/uploads/{dateiname}", response_class=FileResponse)
|
||||||
|
def bild_ausliefern(dateiname: str):
|
||||||
|
"""Liefert ein hochgeladenes Bild aus.
|
||||||
|
|
||||||
|
Der Name kommt aus der URL und ist damit Nutzereingabe - auch wenn wir
|
||||||
|
ihn ursprünglich selbst vergeben haben. Ohne die Musterprüfung liesse
|
||||||
|
sich hier jede Datei des Systems abrufen ("../../etc/passwd").
|
||||||
|
|
||||||
|
Die Bilder liegen bewusst nicht unter static/: sonst wäre jede
|
||||||
|
hochgeladene Datei unter ihrem Namen direkt abrufbar, ohne dass ein
|
||||||
|
fester Content-Type gesetzt werden kann.
|
||||||
|
"""
|
||||||
|
if not images.dateiname_gueltig(dateiname):
|
||||||
|
raise HTTPException(status_code=404, detail="Bild nicht gefunden.")
|
||||||
|
|
||||||
|
pfad = einstellungen().upload_pfad / dateiname
|
||||||
|
if not pfad.is_file():
|
||||||
|
raise HTTPException(status_code=404, detail="Bild nicht gefunden.")
|
||||||
|
|
||||||
|
return FileResponse(
|
||||||
|
pfad,
|
||||||
|
media_type="image/webp",
|
||||||
|
headers={
|
||||||
|
# Fest gesetzt, damit der Browser nichts anderes hineinliest.
|
||||||
|
"X-Content-Type-Options": "nosniff",
|
||||||
|
"Cache-Control": "public, max-age=31536000, immutable",
|
||||||
|
},
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
# --------------------------------------------------------------- Hochladen ---
|
||||||
|
|
||||||
|
@router.post(
|
||||||
|
"/api/v1/items/{item_id}/images",
|
||||||
|
response_model=list[BildAus],
|
||||||
|
status_code=status.HTTP_201_CREATED,
|
||||||
|
tags=["Bilder"],
|
||||||
|
)
|
||||||
|
def bilder_hochladen(
|
||||||
|
item_id: int,
|
||||||
|
dateien: list[UploadFile] = File(...),
|
||||||
|
db: Session = Depends(get_db),
|
||||||
|
):
|
||||||
|
item = crud.item_holen(db, item_id)
|
||||||
|
if item is None:
|
||||||
|
raise HTTPException(status_code=404, detail="Kleidungsstück nicht gefunden.")
|
||||||
|
|
||||||
|
neue: list[ItemBild] = []
|
||||||
|
for datei in dateien:
|
||||||
|
name = _hochladen(datei)
|
||||||
|
neue.append(crud.bild_anhaengen(db, item, name))
|
||||||
|
db.refresh(item)
|
||||||
|
return neue
|
||||||
|
|
||||||
|
|
||||||
|
@router.post(
|
||||||
|
"/api/v1/items/batch-upload",
|
||||||
|
response_model=list[ItemAus],
|
||||||
|
status_code=status.HTTP_201_CREATED,
|
||||||
|
tags=["Bilder"],
|
||||||
|
)
|
||||||
|
def stapel_import(
|
||||||
|
dateien: list[UploadFile] = File(...),
|
||||||
|
db: Session = Depends(get_db),
|
||||||
|
):
|
||||||
|
"""Stapel-Import: pro Foto ein Entwurf.
|
||||||
|
|
||||||
|
Der Weg für den Erstbestand - erst alles mit der Kamera-App
|
||||||
|
fotografieren, dann alle Fotos auf einmal hochladen und die Angaben
|
||||||
|
später nachtragen. Die Entwürfe erscheinen nicht in der Galerie.
|
||||||
|
|
||||||
|
Ein unlesbares Bild lässt den ganzen Stapel scheitern, statt still
|
||||||
|
übersprungen zu werden: bei zwanzig Fotos fiele sonst niemandem auf,
|
||||||
|
dass eines fehlt.
|
||||||
|
"""
|
||||||
|
namen = [_hochladen(datei) for datei in dateien]
|
||||||
|
|
||||||
|
entwuerfe = crud.entwuerfe_anlegen(db, len(namen))
|
||||||
|
for entwurf, name in zip(entwuerfe, namen):
|
||||||
|
crud.bild_anhaengen(db, entwurf, name)
|
||||||
|
db.refresh(entwurf)
|
||||||
|
|
||||||
|
return [
|
||||||
|
ItemAus.model_validate(
|
||||||
|
{**e.__dict__, "display_title": e.anzeige_titel,
|
||||||
|
"kategorie": e.kategorie, "bilder": e.bilder}
|
||||||
|
)
|
||||||
|
for e in entwuerfe
|
||||||
|
]
|
||||||
|
|
||||||
|
|
||||||
|
# ----------------------------------------------------------------- Ändern ---
|
||||||
|
|
||||||
|
@router.post("/api/v1/images/{bild_id}/primary", response_model=BildAus, tags=["Bilder"])
|
||||||
|
def hauptbild_setzen(bild_id: int, db: Session = Depends(get_db)):
|
||||||
|
bild = db.get(ItemBild, bild_id)
|
||||||
|
if bild is None:
|
||||||
|
raise HTTPException(status_code=404, detail="Bild nicht gefunden.")
|
||||||
|
return crud.hauptbild_setzen(db, bild)
|
||||||
|
|
||||||
|
|
||||||
|
@router.delete(
|
||||||
|
"/api/v1/images/{bild_id}",
|
||||||
|
status_code=status.HTTP_204_NO_CONTENT,
|
||||||
|
tags=["Bilder"],
|
||||||
|
)
|
||||||
|
def bild_loeschen(bild_id: int, db: Session = Depends(get_db)):
|
||||||
|
bild = db.get(ItemBild, bild_id)
|
||||||
|
if bild is None:
|
||||||
|
raise HTTPException(status_code=404, detail="Bild nicht gefunden.")
|
||||||
|
dateiname = crud.bild_loeschen(db, bild)
|
||||||
|
images.loeschen(einstellungen().upload_pfad, dateiname)
|
||||||
|
return Response(status_code=status.HTTP_204_NO_CONTENT)
|
||||||
@@ -9,7 +9,8 @@ from fastapi import APIRouter, Depends, HTTPException, Query, Response, status
|
|||||||
from sqlalchemy.exc import IntegrityError
|
from sqlalchemy.exc import IntegrityError
|
||||||
from sqlalchemy.orm import Session
|
from sqlalchemy.orm import Session
|
||||||
|
|
||||||
from app import crud
|
from app import crud, images
|
||||||
|
from app.config import einstellungen
|
||||||
from app.database import get_db
|
from app.database import get_db
|
||||||
from app.models import Geschlecht, Item, Saison, Status
|
from app.models import Geschlecht, Item, Saison, Status
|
||||||
from app.schemas import (
|
from app.schemas import (
|
||||||
@@ -112,9 +113,11 @@ def item_aendern(item_id: int, daten: ItemAendern, db: Session = Depends(get_db)
|
|||||||
@router.delete("/{item_id}", status_code=status.HTTP_204_NO_CONTENT)
|
@router.delete("/{item_id}", status_code=status.HTTP_204_NO_CONTENT)
|
||||||
def item_loeschen(item_id: int, db: Session = Depends(get_db)):
|
def item_loeschen(item_id: int, db: Session = Depends(get_db)):
|
||||||
item = _item_oder_404(db, item_id)
|
item = _item_oder_404(db, item_id)
|
||||||
# Die Dateien selbst löscht Phase 2 (images.py); hier fallen nur die
|
# ON DELETE CASCADE räumt nur die Zeilen in item_images weg. Die Dateien
|
||||||
# Namen an, damit die Aufräumlogik an einer Stelle liegt.
|
# müssen hier weg, sonst läuft das Upload-Verzeichnis mit Waisen voll.
|
||||||
crud.item_loeschen(db, item)
|
dateien = crud.item_loeschen(db, item)
|
||||||
|
for name in dateien:
|
||||||
|
images.loeschen(einstellungen().upload_pfad, name)
|
||||||
return Response(status_code=status.HTTP_204_NO_CONTENT)
|
return Response(status_code=status.HTTP_204_NO_CONTENT)
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
@@ -6,3 +6,8 @@ pydantic>=2.9,<3
|
|||||||
pydantic-settings>=2.6,<3
|
pydantic-settings>=2.6,<3
|
||||||
python-multipart>=0.0.17
|
python-multipart>=0.0.17
|
||||||
jinja2>=3.1,<4
|
jinja2>=3.1,<4
|
||||||
|
pillow>=11.0,<12
|
||||||
|
# iPhones nehmen in HEIC auf. Safari wandelt beim Hochladen über ein
|
||||||
|
# Datei-Feld meist selbst nach JPEG um, aber nicht zuverlässig - und beim
|
||||||
|
# Stapel-Import fiele ein einzelnes nicht lesbares Bild sonst still weg.
|
||||||
|
pillow-heif>=0.20,<1
|
||||||
|
|||||||
@@ -9,11 +9,28 @@ import pytest
|
|||||||
from fastapi.testclient import TestClient
|
from fastapi.testclient import TestClient
|
||||||
from sqlalchemy.orm import sessionmaker
|
from sqlalchemy.orm import sessionmaker
|
||||||
|
|
||||||
|
from app.config import einstellungen
|
||||||
from app.database import Basis, engine_erzeugen, get_db
|
from app.database import Basis, engine_erzeugen, get_db
|
||||||
from app.main import app
|
from app.main import app
|
||||||
from app.seed import stammdaten_anlegen
|
from app.seed import stammdaten_anlegen
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture(autouse=True)
|
||||||
|
def upload_ordner(tmp_path, monkeypatch):
|
||||||
|
"""Jeder Test lädt in ein eigenes Verzeichnis hoch.
|
||||||
|
|
||||||
|
Ohne das schrieben die Tests in das echte uploads/ und liessen dort
|
||||||
|
ihre Bilder liegen. autouse, damit das auch für Tests gilt, die den
|
||||||
|
Ordner nicht selbst anfordern.
|
||||||
|
"""
|
||||||
|
ziel = tmp_path / "uploads"
|
||||||
|
ziel.mkdir()
|
||||||
|
monkeypatch.setenv("UPLOAD_PFAD", str(ziel))
|
||||||
|
einstellungen.cache_clear()
|
||||||
|
yield ziel
|
||||||
|
einstellungen.cache_clear()
|
||||||
|
|
||||||
|
|
||||||
@pytest.fixture
|
@pytest.fixture
|
||||||
def db_sitzung(tmp_path):
|
def db_sitzung(tmp_path):
|
||||||
engine = engine_erzeugen(f"sqlite:///{tmp_path/'test.sqlite'}")
|
engine = engine_erzeugen(f"sqlite:///{tmp_path/'test.sqlite'}")
|
||||||
|
|||||||
@@ -0,0 +1,271 @@
|
|||||||
|
"""Tests zum Bild-Upload.
|
||||||
|
|
||||||
|
Schwerpunkt sind die Fälle, die schiefgehen sollen: der Upload ist die
|
||||||
|
grösste Angriffsfläche der Anwendung.
|
||||||
|
"""
|
||||||
|
|
||||||
|
import io
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
from PIL import Image
|
||||||
|
|
||||||
|
from app import crud
|
||||||
|
|
||||||
|
|
||||||
|
def bild_bytes(breite=800, hoehe=600, format="JPEG", farbe=(200, 30, 30)) -> bytes:
|
||||||
|
puffer = io.BytesIO()
|
||||||
|
Image.new("RGB", (breite, hoehe), farbe).save(puffer, format)
|
||||||
|
return puffer.getvalue()
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture
|
||||||
|
def item_id(client, kategorie_id):
|
||||||
|
return client.post(
|
||||||
|
"/api/v1/items", json={"size": "98/104", "category_id": kategorie_id}
|
||||||
|
).json()["id"]
|
||||||
|
|
||||||
|
|
||||||
|
# ------------------------------------------------------------ Normalfall ---
|
||||||
|
|
||||||
|
def test_hochladen_und_abrufen(client, item_id, upload_ordner):
|
||||||
|
antwort = client.post(
|
||||||
|
f"/api/v1/items/{item_id}/images",
|
||||||
|
files={"dateien": ("foto.jpg", bild_bytes(), "image/jpeg")},
|
||||||
|
)
|
||||||
|
assert antwort.status_code == 201
|
||||||
|
bilder = antwort.json()
|
||||||
|
assert len(bilder) == 1
|
||||||
|
name = bilder[0]["image_url"]
|
||||||
|
|
||||||
|
# Selbst vergebener Name, nicht der hochgeladene
|
||||||
|
assert name.endswith(".webp")
|
||||||
|
assert "foto" not in name
|
||||||
|
|
||||||
|
# Als WebP abgelegt, nicht als JPEG
|
||||||
|
datei = upload_ordner / name
|
||||||
|
assert datei.is_file()
|
||||||
|
assert Image.open(datei).format == "WEBP"
|
||||||
|
|
||||||
|
ausgeliefert = client.get(f"/uploads/{name}")
|
||||||
|
assert ausgeliefert.status_code == 200
|
||||||
|
assert ausgeliefert.headers["content-type"] == "image/webp"
|
||||||
|
assert ausgeliefert.headers["x-content-type-options"] == "nosniff"
|
||||||
|
|
||||||
|
|
||||||
|
def test_erstes_bild_wird_hauptbild(client, item_id):
|
||||||
|
erstes = client.post(
|
||||||
|
f"/api/v1/items/{item_id}/images",
|
||||||
|
files={"dateien": ("a.jpg", bild_bytes(), "image/jpeg")},
|
||||||
|
).json()[0]
|
||||||
|
zweites = client.post(
|
||||||
|
f"/api/v1/items/{item_id}/images",
|
||||||
|
files={"dateien": ("b.jpg", bild_bytes(), "image/jpeg")},
|
||||||
|
).json()[0]
|
||||||
|
assert erstes["is_primary"] is True
|
||||||
|
assert zweites["is_primary"] is False
|
||||||
|
|
||||||
|
|
||||||
|
def test_hauptbild_umstellen(client, item_id):
|
||||||
|
client.post(f"/api/v1/items/{item_id}/images",
|
||||||
|
files={"dateien": ("a.jpg", bild_bytes(), "image/jpeg")})
|
||||||
|
zweites = client.post(f"/api/v1/items/{item_id}/images",
|
||||||
|
files={"dateien": ("b.jpg", bild_bytes(), "image/jpeg")}).json()[0]
|
||||||
|
|
||||||
|
client.post(f"/api/v1/images/{zweites['id']}/primary")
|
||||||
|
bilder = client.get(f"/api/v1/items/{item_id}").json()["bilder"]
|
||||||
|
assert [b["is_primary"] for b in bilder] == [False, True]
|
||||||
|
|
||||||
|
|
||||||
|
def test_grosses_bild_wird_verkleinert(client, item_id, upload_ordner):
|
||||||
|
bild = client.post(
|
||||||
|
f"/api/v1/items/{item_id}/images",
|
||||||
|
files={"dateien": ("gross.jpg", bild_bytes(4000, 3000), "image/jpeg")},
|
||||||
|
).json()[0]
|
||||||
|
assert Image.open(upload_ordner / bild["image_url"]).width == 1200
|
||||||
|
|
||||||
|
|
||||||
|
def test_kleines_bild_bleibt_wie_es_ist(client, item_id, upload_ordner):
|
||||||
|
bild = client.post(
|
||||||
|
f"/api/v1/items/{item_id}/images",
|
||||||
|
files={"dateien": ("klein.jpg", bild_bytes(400, 300), "image/jpeg")},
|
||||||
|
).json()[0]
|
||||||
|
assert Image.open(upload_ordner / bild["image_url"]).width == 400
|
||||||
|
|
||||||
|
|
||||||
|
# --------------------------------------------------------- Abwehrverhalten ---
|
||||||
|
|
||||||
|
def test_getarnte_datei_wird_abgewiesen(client, item_id):
|
||||||
|
"""Eine PHP-Datei mit der Endung .jpg darf nicht durchkommen - der
|
||||||
|
Inhalt entscheidet, nicht die Endung."""
|
||||||
|
antwort = client.post(
|
||||||
|
f"/api/v1/items/{item_id}/images",
|
||||||
|
files={"dateien": ("harmlos.jpg", b"<?php system($_GET['c']); ?>", "image/jpeg")},
|
||||||
|
)
|
||||||
|
assert antwort.status_code == 422
|
||||||
|
|
||||||
|
|
||||||
|
def test_leere_datei_wird_abgewiesen(client, item_id):
|
||||||
|
antwort = client.post(
|
||||||
|
f"/api/v1/items/{item_id}/images",
|
||||||
|
files={"dateien": ("leer.jpg", b"", "image/jpeg")},
|
||||||
|
)
|
||||||
|
assert antwort.status_code == 422
|
||||||
|
|
||||||
|
|
||||||
|
def test_zu_grosse_datei_wird_abgewiesen(client, item_id, monkeypatch):
|
||||||
|
from app.config import einstellungen
|
||||||
|
|
||||||
|
einstellungen.cache_clear()
|
||||||
|
monkeypatch.setenv("MAX_UPLOAD_BYTES", "1024")
|
||||||
|
try:
|
||||||
|
antwort = client.post(
|
||||||
|
f"/api/v1/items/{item_id}/images",
|
||||||
|
files={"dateien": ("gross.jpg", bild_bytes(2000, 2000), "image/jpeg")},
|
||||||
|
)
|
||||||
|
assert antwort.status_code == 422
|
||||||
|
finally:
|
||||||
|
einstellungen.cache_clear()
|
||||||
|
|
||||||
|
|
||||||
|
def test_dekompressionsbombe_wird_abgewiesen(client, item_id):
|
||||||
|
"""Ein paar hundert Kilobyte, die beim Entpacken den Speicher füllen.
|
||||||
|
|
||||||
|
Pillow allein würde hier nur warnen - der Fehler kommt erst bei der
|
||||||
|
doppelten Pixelzahl. Darum die ausdrückliche Prüfung in images.py.
|
||||||
|
"""
|
||||||
|
puffer = io.BytesIO()
|
||||||
|
# 20000 x 20000 = 400 Megapixel, als einfarbiges PNG winzig komprimiert
|
||||||
|
Image.new("RGB", (20000, 20000), (0, 0, 0)).save(puffer, "PNG", compress_level=9)
|
||||||
|
roh = puffer.getvalue()
|
||||||
|
assert len(roh) < 2 * 1024 * 1024, "Testbild sollte klein sein"
|
||||||
|
|
||||||
|
antwort = client.post(
|
||||||
|
f"/api/v1/items/{item_id}/images",
|
||||||
|
files={"dateien": ("bombe.png", roh, "image/png")},
|
||||||
|
)
|
||||||
|
assert antwort.status_code == 422
|
||||||
|
assert "Bildpunkte" in antwort.json()["detail"]
|
||||||
|
|
||||||
|
|
||||||
|
def test_pfadmanipulation_beim_ausliefern(client):
|
||||||
|
"""Der Dateiname kommt aus der URL und ist damit Nutzereingabe."""
|
||||||
|
for versuch in ["../../etc/passwd", "..%2f..%2fetc%2fpasswd", "beliebig.txt",
|
||||||
|
"0123456789abcdef.webp"]:
|
||||||
|
assert client.get(f"/uploads/{versuch}").status_code == 404
|
||||||
|
|
||||||
|
|
||||||
|
def test_heic_vom_iphone_wird_angenommen(client, item_id, upload_ordner):
|
||||||
|
"""iPhones nehmen in HEIC auf. Pillow allein kann das nicht öffnen -
|
||||||
|
ohne pillow-heif fiele beim Stapel-Import jedes iPhone-Foto durch."""
|
||||||
|
puffer = io.BytesIO()
|
||||||
|
Image.new("RGB", (2000, 1500), (40, 90, 160)).save(puffer, "HEIF")
|
||||||
|
|
||||||
|
antwort = client.post(
|
||||||
|
f"/api/v1/items/{item_id}/images",
|
||||||
|
files={"dateien": ("IMG_0042.HEIC", puffer.getvalue(), "image/heic")},
|
||||||
|
)
|
||||||
|
assert antwort.status_code == 201
|
||||||
|
gespeichert = Image.open(upload_ordner / antwort.json()[0]["image_url"])
|
||||||
|
assert gespeichert.format == "WEBP"
|
||||||
|
|
||||||
|
|
||||||
|
def test_exif_wird_entfernt(client, item_id, upload_ordner):
|
||||||
|
"""Handyfotos tragen oft die GPS-Koordinaten der Wohnung mit sich."""
|
||||||
|
puffer = io.BytesIO()
|
||||||
|
bild = Image.new("RGB", (600, 400), (10, 120, 10))
|
||||||
|
exif = bild.getexif()
|
||||||
|
exif[271] = "TestKamera" # Make
|
||||||
|
exif[274] = 1 # Orientation
|
||||||
|
bild.save(puffer, "JPEG", exif=exif)
|
||||||
|
|
||||||
|
hochgeladen = client.post(
|
||||||
|
f"/api/v1/items/{item_id}/images",
|
||||||
|
files={"dateien": ("mit-exif.jpg", puffer.getvalue(), "image/jpeg")},
|
||||||
|
).json()[0]
|
||||||
|
|
||||||
|
gespeichert = Image.open(upload_ordner / hochgeladen["image_url"])
|
||||||
|
assert not dict(gespeichert.getexif())
|
||||||
|
|
||||||
|
|
||||||
|
# -------------------------------------------------------------- Aufräumen ---
|
||||||
|
|
||||||
|
def test_bild_loeschen_entfernt_die_datei(client, item_id, upload_ordner):
|
||||||
|
bild = client.post(
|
||||||
|
f"/api/v1/items/{item_id}/images",
|
||||||
|
files={"dateien": ("a.jpg", bild_bytes(), "image/jpeg")},
|
||||||
|
).json()[0]
|
||||||
|
datei = upload_ordner / bild["image_url"]
|
||||||
|
assert datei.is_file()
|
||||||
|
|
||||||
|
assert client.delete(f"/api/v1/images/{bild['id']}").status_code == 204
|
||||||
|
assert not datei.exists()
|
||||||
|
|
||||||
|
|
||||||
|
def test_naechstes_bild_rueckt_als_hauptbild_nach(client, item_id):
|
||||||
|
erstes = client.post(f"/api/v1/items/{item_id}/images",
|
||||||
|
files={"dateien": ("a.jpg", bild_bytes(), "image/jpeg")}).json()[0]
|
||||||
|
client.post(f"/api/v1/items/{item_id}/images",
|
||||||
|
files={"dateien": ("b.jpg", bild_bytes(), "image/jpeg")})
|
||||||
|
|
||||||
|
client.delete(f"/api/v1/images/{erstes['id']}")
|
||||||
|
bilder = client.get(f"/api/v1/items/{item_id}").json()["bilder"]
|
||||||
|
assert len(bilder) == 1
|
||||||
|
assert bilder[0]["is_primary"] is True
|
||||||
|
|
||||||
|
|
||||||
|
def test_eintrag_loeschen_entfernt_die_dateien(client, item_id, upload_ordner):
|
||||||
|
"""ON DELETE CASCADE räumt nur die Zeilen weg, nicht die Dateien."""
|
||||||
|
bilder = client.post(
|
||||||
|
f"/api/v1/items/{item_id}/images",
|
||||||
|
files=[("dateien", ("a.jpg", bild_bytes(), "image/jpeg")),
|
||||||
|
("dateien", ("b.jpg", bild_bytes(), "image/jpeg"))],
|
||||||
|
).json()
|
||||||
|
dateien = [upload_ordner / b["image_url"] for b in bilder]
|
||||||
|
assert all(d.is_file() for d in dateien)
|
||||||
|
|
||||||
|
client.delete(f"/api/v1/items/{item_id}")
|
||||||
|
assert not any(d.exists() for d in dateien)
|
||||||
|
|
||||||
|
|
||||||
|
# ----------------------------------------------------------- Stapel-Import ---
|
||||||
|
|
||||||
|
def test_stapel_import_legt_entwuerfe_an(client, upload_ordner):
|
||||||
|
antwort = client.post(
|
||||||
|
"/api/v1/items/batch-upload",
|
||||||
|
files=[("dateien", (f"foto{i}.jpg", bild_bytes(), "image/jpeg")) for i in range(5)],
|
||||||
|
)
|
||||||
|
assert antwort.status_code == 201
|
||||||
|
entwuerfe = antwort.json()
|
||||||
|
assert len(entwuerfe) == 5
|
||||||
|
assert all(e["status"] == "draft" for e in entwuerfe)
|
||||||
|
assert all(len(e["bilder"]) == 1 for e in entwuerfe)
|
||||||
|
assert all(e["display_title"] == "Kleidungsstück" for e in entwuerfe)
|
||||||
|
|
||||||
|
# Nicht in der Galerie, aber als Entwurf auffindbar
|
||||||
|
assert client.get("/api/v1/items").json()["total"] == 0
|
||||||
|
assert client.get("/api/v1/items?status=draft").json()["total"] == 5
|
||||||
|
|
||||||
|
|
||||||
|
def test_stapel_import_bricht_bei_kaputtem_bild_ab(client):
|
||||||
|
"""Lieber der ganze Stapel als ein still fehlendes Foto unter zwanzig."""
|
||||||
|
antwort = client.post(
|
||||||
|
"/api/v1/items/batch-upload",
|
||||||
|
files=[("dateien", ("gut.jpg", bild_bytes(), "image/jpeg")),
|
||||||
|
("dateien", ("kaputt.jpg", b"kein bild", "image/jpeg"))],
|
||||||
|
)
|
||||||
|
assert antwort.status_code == 422
|
||||||
|
|
||||||
|
|
||||||
|
def test_entwurf_wird_mit_bild_zum_eintrag(client, kategorie_id):
|
||||||
|
entwurf = client.post(
|
||||||
|
"/api/v1/items/batch-upload",
|
||||||
|
files={"dateien": ("foto.jpg", bild_bytes(), "image/jpeg")},
|
||||||
|
).json()[0]
|
||||||
|
|
||||||
|
fertig = client.patch(
|
||||||
|
f"/api/v1/items/{entwurf['id']}",
|
||||||
|
json={"size": "92", "category_id": kategorie_id, "status": "available"},
|
||||||
|
).json()
|
||||||
|
assert fertig["status"] == "available"
|
||||||
|
assert len(fertig["bilder"]) == 1
|
||||||
|
assert fertig["display_title"].endswith("92")
|
||||||
Reference in New Issue
Block a user