patrickandClaude Sonnet 5
707701d897
Duplikat-Erkennung: SHA-256-Hash je Dokument (Migration 0024), Upload wird mit 409 abgelehnt, wenn dieselbe Datei bereits an derselben Entität liegt (Vergleich bewusst pro Entität, nicht global - gleiche Datei an zwei Objekten ist kein Duplikat). Frontend zeigt die Backend-Fehlermeldung (Dateiname/Datum des bestehenden Dokuments) statt generischem Text. Nebenbei gefunden+gefixt: "Ansehen" (PDF/JPG/PNG online statt Download) funktionierte nicht zuverlässig, weil window.open() erst nach einem await fetch() aufgerufen wurde - Popup-Blocker werten das nicht mehr als direkte Nutzeraktion. Jetzt öffnet der Tab sofort synchron, die Blob-URL wird nachgeladen. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01CVgbozhYmuEhiEJHffRXCV
122 lines
3.6 KiB
Python
122 lines
3.6 KiB
Python
import hashlib
|
|
import os
|
|
import uuid
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.core.app_settings import settings
|
|
from app.models.dokument import Dokument
|
|
|
|
# Erlaubte MIME-Types (Nutzer-Vorgabe Modul 10: PDF/Bilder/Prüfprotokolle/
|
|
# Wartungsberichte/Bedienungsanleitungen/Rechnungen/Zulassungsdokumente) -
|
|
# Whitelist statt Blacklist (owasp-Grundsatz: Datei-Upload ist klassischer
|
|
# Angriffsvektor, z.B. .html/.svg mit eingebettetem Skript).
|
|
ERLAUBTE_MIME_TYPES = {
|
|
"application/pdf",
|
|
"image/jpeg",
|
|
"image/png",
|
|
"image/webp",
|
|
}
|
|
|
|
|
|
class DateityperlaubtError(Exception):
|
|
pass
|
|
|
|
|
|
class DateizugrossError(Exception):
|
|
pass
|
|
|
|
|
|
class DokumentDuplikatError(Exception):
|
|
"""DOC-001: identische Datei (SHA-256) bereits an dieser Entität vorhanden."""
|
|
|
|
def __init__(self, bestehendes_dokument: Dokument):
|
|
self.bestehendes_dokument = bestehendes_dokument
|
|
super().__init__(f"Duplikat von Dokument {bestehendes_dokument.id}")
|
|
|
|
|
|
def _upload_pfad() -> Path:
|
|
pfad = Path(settings.upload_dir)
|
|
pfad.mkdir(parents=True, exist_ok=True)
|
|
return pfad
|
|
|
|
|
|
async def speichere_dokument(
|
|
db: AsyncSession,
|
|
*,
|
|
entitaet_typ: str,
|
|
entitaet_id: str,
|
|
dateiname: str,
|
|
mime_type: str,
|
|
inhalt: bytes,
|
|
beschreibung: str | None,
|
|
hochgeladen_von: int,
|
|
) -> Dokument:
|
|
if mime_type not in ERLAUBTE_MIME_TYPES:
|
|
raise DateityperlaubtError(mime_type)
|
|
if len(inhalt) > settings.max_upload_size_mb * 1024 * 1024:
|
|
raise DateizugrossError(len(inhalt))
|
|
|
|
dateihash = hashlib.sha256(inhalt).hexdigest()
|
|
bestehendes = await db.execute(
|
|
select(Dokument).where(
|
|
Dokument.entitaet_typ == entitaet_typ,
|
|
Dokument.entitaet_id == entitaet_id,
|
|
Dokument.dateihash == dateihash,
|
|
)
|
|
)
|
|
duplikat = bestehendes.scalars().first()
|
|
if duplikat is not None:
|
|
raise DokumentDuplikatError(duplikat)
|
|
|
|
# Speichername ist server-generiert (UUID), NIEMALS der Original-Dateiname -
|
|
# verhindert Path-Traversal (../../etc/passwd) und Namenskollisionen.
|
|
endung = Path(dateiname).suffix[:10]
|
|
speichername = f"{uuid.uuid4()}{endung}"
|
|
ziel = _upload_pfad() / speichername
|
|
ziel.write_bytes(inhalt)
|
|
|
|
dokument = Dokument(
|
|
entitaet_typ=entitaet_typ,
|
|
entitaet_id=entitaet_id,
|
|
dateiname=dateiname,
|
|
speicherpfad=speichername,
|
|
mime_type=mime_type,
|
|
groesse_bytes=len(inhalt),
|
|
dateihash=dateihash,
|
|
beschreibung=beschreibung,
|
|
hochgeladen_von=hochgeladen_von,
|
|
hochgeladen_am=datetime.now(timezone.utc),
|
|
)
|
|
db.add(dokument)
|
|
await db.flush()
|
|
return dokument
|
|
|
|
|
|
async def liste_fuer_entitaet(db: AsyncSession, *, entitaet_typ: str, entitaet_id: str) -> list[Dokument]:
|
|
result = await db.execute(
|
|
select(Dokument)
|
|
.where(Dokument.entitaet_typ == entitaet_typ, Dokument.entitaet_id == entitaet_id)
|
|
.order_by(Dokument.hochgeladen_am.desc())
|
|
)
|
|
return list(result.scalars().all())
|
|
|
|
|
|
def dateipfad(dokument: Dokument) -> Path:
|
|
return _upload_pfad() / dokument.speicherpfad
|
|
|
|
|
|
async def loesche_dokument(db: AsyncSession, *, dokument: Dokument) -> None:
|
|
pfad = dateipfad(dokument)
|
|
await db.delete(dokument)
|
|
await db.flush()
|
|
# Datei erst nach erfolgreichem DB-Commit-Vorbereiten löschen (flush wirft
|
|
# bei FK-Problemen, bevor die Datei weg ist) - hier gibt es keine
|
|
# eingehenden FKs auf dokument, daher unkritisch, aber Reihenfolge bewusst
|
|
# gewählt für den Fall künftiger Referenzen.
|
|
if pfad.exists():
|
|
os.remove(pfad)
|