chore: initial import
This commit is contained in:
@@ -0,0 +1,275 @@
|
||||
# -*- 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,
|
||||
}
|
||||
Reference in New Issue
Block a user