276 lines
10 KiB
Python
276 lines
10 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""
|
|
Plugins als ZIP hochladen.
|
|
|
|
Identität eines Plugins ist die ``id`` aus ``plugin.json`` — nicht der
|
|
Ordnername. Daran hängen Datenordner (``data/<id>/``), Adresse
|
|
(``/plugins/<id>``), Modul-Namensraum und Job-Zuordnung. Wird ein Plugin mit
|
|
bekannter ``id`` hochgeladen, ersetzt es genau dessen Ordner, egal wie der
|
|
hochgeladene Ordner heißt.
|
|
|
|
Grundregel beim Ersetzen: **nur ergänzen und überschreiben, nie löschen.**
|
|
Dateien, die im Archiv nicht vorkommen, bleiben unangetastet. Der Datenordner
|
|
liegt ohnehin außerhalb des Plugin-Ordners und wird nie berührt.
|
|
|
|
Sicherheit: Ein Plugin ist ausführbarer Code, den der Host importiert. Dieser
|
|
Import ist damit bewusst eine Administrator-Funktion. Geprüft wird trotzdem
|
|
alles, was sich prüfen lässt — Pfadausbrüche, Symlinks, Größen, Bomben.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import io
|
|
import json
|
|
import re
|
|
import shutil
|
|
import stat
|
|
import zipfile
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
# Was eine gültige Plugin-Kennung sein darf. Bewusst eng: sie wird zum
|
|
# Ordnernamen, zum URL-Bestandteil und zum Modulnamen.
|
|
ID_MUSTER = re.compile(r"^[a-z0-9][a-z0-9._-]{0,63}$")
|
|
|
|
MAX_ARCHIV = 25 * 1024 * 1024 # 25 MB gepackt
|
|
MAX_ENTPACKT = 120 * 1024 * 1024 # 120 MB entpackt (gegen ZIP-Bomben)
|
|
MAX_EINTRAEGE = 3000
|
|
|
|
# Diese Namen dürfen nie aus einem Archiv geschrieben werden.
|
|
VERBOTEN = {".git", ".ssh", "__pycache__"}
|
|
|
|
|
|
class ImportFehler(ValueError):
|
|
"""Fachlicher Fehler mit einer Meldung, die man dem Menschen zeigen kann."""
|
|
|
|
|
|
def version_tupel(text: str) -> tuple:
|
|
"""„1.10.2“ → (1, 10, 2). Nicht-Zahlen landen hinten und zählen als 0."""
|
|
teile = re.split(r"[.\-+]", str(text or "0"))
|
|
zahlen = []
|
|
for t in teile:
|
|
m = re.match(r"^(\d+)", t)
|
|
zahlen.append(int(m.group(1)) if m else 0)
|
|
while len(zahlen) < 3:
|
|
zahlen.append(0)
|
|
return tuple(zahlen[:4])
|
|
|
|
|
|
def _pfad_ok(name: str) -> bool:
|
|
"""Verhindert Pfadausbrüche („zip slip“) und absolute Pfade."""
|
|
if not name or name.startswith("/") or name.startswith("\\"):
|
|
return False
|
|
if ":" in name.split("/")[0] and len(name.split("/")[0]) == 2:
|
|
return False # Laufwerksbuchstabe C:
|
|
teile = name.replace("\\", "/").split("/")
|
|
for t in teile:
|
|
if t in ("", ".", ".."):
|
|
return False
|
|
if t in VERBOTEN:
|
|
return False
|
|
return True
|
|
|
|
|
|
def _ist_symlink(info: zipfile.ZipInfo) -> bool:
|
|
return stat.S_ISLNK(info.external_attr >> 16)
|
|
|
|
|
|
def lies_archiv(daten: bytes) -> dict[str, Any]:
|
|
"""
|
|
Prüft das Archiv und liest die ``plugin.json``, ohne irgendetwas zu
|
|
schreiben. Liefert die Beschreibung des Plugins und die Liste der Dateien.
|
|
"""
|
|
if len(daten) > MAX_ARCHIV:
|
|
raise ImportFehler(
|
|
f"Archiv ist zu groß ({len(daten)//1024//1024} MB, erlaubt sind "
|
|
f"{MAX_ARCHIV//1024//1024} MB).")
|
|
try:
|
|
zf = zipfile.ZipFile(io.BytesIO(daten))
|
|
except zipfile.BadZipFile:
|
|
raise ImportFehler("Das ist keine lesbare ZIP-Datei.")
|
|
|
|
infos = [i for i in zf.infolist() if not i.is_dir()]
|
|
if not infos:
|
|
raise ImportFehler("Das Archiv ist leer.")
|
|
if len(infos) > MAX_EINTRAEGE:
|
|
raise ImportFehler(f"Zu viele Dateien im Archiv ({len(infos)}).")
|
|
|
|
gesamt = 0
|
|
for i in infos:
|
|
if _ist_symlink(i):
|
|
raise ImportFehler(f"Symbolische Verknüpfungen sind nicht erlaubt: {i.filename}")
|
|
if not _pfad_ok(i.filename):
|
|
raise ImportFehler(f"Unerlaubter Pfad im Archiv: {i.filename}")
|
|
gesamt += i.file_size
|
|
if gesamt > MAX_ENTPACKT:
|
|
raise ImportFehler("Das Archiv wäre entpackt zu groß.")
|
|
|
|
# plugin.json mit dem kürzesten Pfad bestimmt den Wurzelordner.
|
|
kandidaten = [i.filename for i in infos
|
|
if i.filename.replace("\\", "/").split("/")[-1] == "plugin.json"]
|
|
if not kandidaten:
|
|
raise ImportFehler("Im Archiv fehlt eine plugin.json.")
|
|
kandidaten.sort(key=lambda n: (n.count("/"), len(n)))
|
|
manifest_pfad = kandidaten[0]
|
|
wurzel = manifest_pfad.rsplit("/", 1)[0] + "/" if "/" in manifest_pfad else ""
|
|
|
|
try:
|
|
roh = json.loads(zf.read(manifest_pfad).decode("utf-8"))
|
|
except (UnicodeDecodeError, json.JSONDecodeError) as exc:
|
|
raise ImportFehler(f"plugin.json ist nicht lesbar: {exc}")
|
|
if not isinstance(roh, dict):
|
|
raise ImportFehler("plugin.json muss ein Objekt enthalten.")
|
|
|
|
pid = str(roh.get("id") or "").strip()
|
|
if not pid:
|
|
raise ImportFehler("In der plugin.json fehlt die „id“ — daran wird das "
|
|
"Plugin erkannt.")
|
|
if not ID_MUSTER.match(pid):
|
|
raise ImportFehler(
|
|
f"Die id „{pid}“ ist nicht zulässig. Erlaubt sind Kleinbuchstaben, "
|
|
"Ziffern, Punkt, Bindestrich und Unterstrich (Beginn mit Buchstabe "
|
|
"oder Ziffer).")
|
|
|
|
dateien = [i.filename for i in infos if i.filename.startswith(wurzel)]
|
|
if not any(f[len(wurzel):] == "backend.py" for f in dateien) \
|
|
and not roh.get("entrypoint"):
|
|
raise ImportFehler("Im Archiv fehlt die backend.py.")
|
|
|
|
return {
|
|
"zf": zf,
|
|
"wurzel": wurzel,
|
|
"dateien": dateien,
|
|
"manifest": roh,
|
|
"id": pid,
|
|
"name": str(roh.get("name") or pid),
|
|
"version": str(roh.get("version") or "0.0.0"),
|
|
}
|
|
|
|
|
|
def finde_vorhandenes(plugins_dir: Path, pid: str) -> Path | None:
|
|
"""
|
|
Sucht den Ordner eines bereits vorhandenen Plugins **anhand der id**, nicht
|
|
anhand des Ordnernamens. Nur so wird ein umbenanntes Plugin sauber ersetzt
|
|
statt doppelt angelegt.
|
|
"""
|
|
for manifest in sorted(plugins_dir.glob("*/plugin.json")):
|
|
try:
|
|
roh = json.loads(manifest.read_text(encoding="utf-8"))
|
|
except (OSError, json.JSONDecodeError):
|
|
continue
|
|
vorhandene_id = str(roh.get("id") or manifest.parent.name).strip()
|
|
if vorhandene_id == pid:
|
|
return manifest.parent
|
|
return None
|
|
|
|
|
|
def pruefe(daten: bytes, plugins_dir: Path) -> dict[str, Any]:
|
|
"""Vorschau: Was würde der Import tun? Schreibt nichts."""
|
|
a = lies_archiv(daten)
|
|
vorhanden = finde_vorhandenes(plugins_dir, a["id"])
|
|
alt_version = None
|
|
if vorhanden:
|
|
try:
|
|
alt = json.loads((vorhanden / "plugin.json").read_text(encoding="utf-8"))
|
|
alt_version = str(alt.get("version") or "0.0.0")
|
|
except (OSError, json.JSONDecodeError):
|
|
alt_version = "?"
|
|
|
|
neu = version_tupel(a["version"])
|
|
alt_t = version_tupel(alt_version) if alt_version else None
|
|
if not vorhanden:
|
|
art, warnung = "neu", None
|
|
elif alt_t is None or neu > alt_t:
|
|
art, warnung = "aktualisierung", None
|
|
elif neu == alt_t:
|
|
art = "gleiche_version"
|
|
warnung = (f"Version {a['version']} ist bereits installiert. Beim "
|
|
"Fortfahren werden die Dateien überschrieben.")
|
|
else:
|
|
art = "aelter"
|
|
warnung = (f"Die hochgeladene Version {a['version']} ist ÄLTER als die "
|
|
f"installierte {alt_version}.")
|
|
|
|
return {
|
|
"id": a["id"],
|
|
"name": a["name"],
|
|
"version": a["version"],
|
|
"version_alt": alt_version,
|
|
"art": art,
|
|
"warnung": warnung,
|
|
"ordner": vorhanden.name if vorhanden else a["id"],
|
|
"dateien": len(a["dateien"]),
|
|
"bestaetigung_noetig": art in ("gleiche_version", "aelter"),
|
|
}
|
|
|
|
|
|
def importiere(daten: bytes, plugins_dir: Path, sicherungen: Path,
|
|
force: bool = False) -> dict[str, Any]:
|
|
"""
|
|
Schreibt das Plugin nach ``plugins/``.
|
|
|
|
* Vorhandenes wird **anhand der id** gefunden und im selben Ordner ergänzt.
|
|
* Es wird nur geschrieben und überschrieben — **nichts gelöscht**. Dateien,
|
|
die das Archiv nicht enthält, bleiben liegen.
|
|
* Vorher wandert der bisherige Stand vollständig in eine Sicherung.
|
|
"""
|
|
vorschau = pruefe(daten, plugins_dir)
|
|
if vorschau["bestaetigung_noetig"] and not force:
|
|
raise ImportFehler(vorschau["warnung"] or "Bestätigung nötig.")
|
|
|
|
a = lies_archiv(daten)
|
|
zf, wurzel = a["zf"], a["wurzel"]
|
|
pid = a["id"]
|
|
|
|
vorhanden = finde_vorhandenes(plugins_dir, pid)
|
|
ziel = vorhanden if vorhanden else (plugins_dir / pid)
|
|
ziel_aufgeloest = ziel.resolve()
|
|
plugins_aufgeloest = plugins_dir.resolve()
|
|
if plugins_aufgeloest not in ziel_aufgeloest.parents:
|
|
raise ImportFehler("Zielordner liegt außerhalb von plugins/.")
|
|
|
|
# Sicherung des bisherigen Standes (macht den Import umkehrbar).
|
|
sicherung = None
|
|
if ziel.is_dir():
|
|
stempel = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
|
|
basis = sicherungen / pid
|
|
basis.mkdir(parents=True, exist_ok=True)
|
|
sicherung = basis / stempel
|
|
# Mehrere Importe in derselben Sekunde dürfen sich nicht überschreiben.
|
|
zaehler = 2
|
|
while sicherung.exists():
|
|
sicherung = basis / f"{stempel}-{zaehler}"
|
|
zaehler += 1
|
|
shutil.copytree(ziel, sicherung)
|
|
|
|
ziel.mkdir(parents=True, exist_ok=True)
|
|
geschrieben, uebersprungen = 0, 0
|
|
for name in a["dateien"]:
|
|
rel = name[len(wurzel):]
|
|
if not rel or rel.endswith("/"):
|
|
continue
|
|
if not _pfad_ok(rel):
|
|
uebersprungen += 1
|
|
continue
|
|
pfad = (ziel / rel).resolve()
|
|
if ziel_aufgeloest != pfad and ziel_aufgeloest not in pfad.parents:
|
|
uebersprungen += 1 # Ausbruchsversuch — überspringen
|
|
continue
|
|
pfad.parent.mkdir(parents=True, exist_ok=True)
|
|
with zf.open(name) as quelle, open(pfad, "wb") as ablage:
|
|
shutil.copyfileobj(quelle, ablage, length=64 * 1024)
|
|
geschrieben += 1
|
|
|
|
return {
|
|
"ok": True,
|
|
"id": pid,
|
|
"name": a["name"],
|
|
"version": a["version"],
|
|
"version_alt": vorschau["version_alt"],
|
|
"art": vorschau["art"],
|
|
"ordner": ziel.name,
|
|
"geschrieben": geschrieben,
|
|
"uebersprungen": uebersprungen,
|
|
"sicherung": str(sicherung) if sicherung else None,
|
|
}
|