Aether/scripts/upload-abbruch-selbsttest.py

#!/usr/bin/env python3
"""Selbsttest: Was bleibt liegen, wenn ein Upload abgebrochen wird?

    python3 scripts/upload-abbruch-selbsttest.py [BASIS-URL] [BENUTZER] [PASSWORT]

Die Oberfläche hat jetzt ein Kreuz, das einen laufenden Upload fallen lässt.
Die Behauptung dahinter lautet: **Auf dem Server bleibt dabei nichts zurück.**
Sie stützt sich darauf, dass der Speicher in eine Temp-Datei schreibt und erst
am Ende umbenennt – aber eine Bauweise ist keine Prüfung. Hier wird sie eine.

Geprüft wird gegen die **laufende Instanz**, mit einer echten, mitten im Strom
abgeschnittenen Verbindung:

  * kein Eintrag in der Datenbank
  * keine Bytes am Zielpfad
  * keine liegengebliebene Temp-Datei
  * und der gefährlichste Fall: Wird eine **vorhandene** Datei überschrieben
    und der Upload dabei abgebrochen, muss die alte Fassung unversehrt bleiben.
    Genau dort verschiebt der Server sie kurzzeitig zur Seite.

Alles Angelegte wird am Ende wieder entfernt.
"""
import hashlib
import http.cookiejar
import json
import socket
import sys
import time
import urllib.error
import urllib.parse
import urllib.request
import uuid

BASE = (sys.argv[1] if len(sys.argv) > 1 else "http://localhost:8090").rstrip("/")
USER = sys.argv[2] if len(sys.argv) > 2 else "admin"
PASSWORT = sys.argv[3] if len(sys.argv) > 3 else "admin123"

ok_count = 0
fail = []


def check(name, cond, detail=""):
    global ok_count
    if cond:
        ok_count += 1
        print(f"  \033[32mok\033[0m   {name}")
    else:
        fail.append(name)
        print(f"  \033[31mFEHL\033[0m {name}   {detail}")


def sitzung():
    jar = http.cookiejar.CookieJar()
    return urllib.request.build_opener(urllib.request.HTTPCookieProcessor(jar)), jar


def ruf(opener, method, path, body=None, headers=None):
    hdrs = dict(headers or {})
    data = None
    if body is not None:
        data = json.dumps(body).encode()
        hdrs["Content-Type"] = "application/json"
    req = urllib.request.Request(BASE + path, data=data, headers=hdrs, method=method)
    try:
        with opener.open(req) as r:
            text = r.read().decode()
            return r.status, (json.loads(text).get("data") if text else None)
    except urllib.error.HTTPError as e:
        return e.code, None


def anmelden():
    o, jar = sitzung()
    st, d = ruf(o, "POST", "/api/v1/auth/login", {"username": USER, "password": PASSWORT})
    assert st == 200, f"Anmeldung fehlgeschlagen: {st}"
    return o, jar, {"X-CSRF-Token": d["csrf_token"]}


def keksleiste(jar):
    return "; ".join(f"{c.name}={c.value}" for c in jar)


def hochladen(opener, csrf, name, inhalt: bytes, folder_id=None):
    grenze = "----aether" + uuid.uuid4().hex
    teile = []
    if folder_id:
        teile.append(f'--{grenze}\r\nContent-Disposition: form-data; name="folder_id"\r\n\r\n{folder_id}\r\n')
    kopf = (f'--{grenze}\r\nContent-Disposition: form-data; name="file"; filename="{name}"\r\n'
            f'Content-Type: application/octet-stream\r\n\r\n')
    leib = b"".join(t.encode() for t in teile) + kopf.encode() + inhalt + f"\r\n--{grenze}--\r\n".encode()
    req = urllib.request.Request(BASE + "/api/v1/files", data=leib, method="POST",
                                 headers={"Content-Type": f"multipart/form-data; boundary={grenze}", **csrf})
    with opener.open(req) as r:
        return json.loads(r.read().decode())["data"]


def abgebrochen_hochladen(jar, csrf, name, folder_id, gesamt, nach_bytes):
    """Baut die Anfrage von Hand auf einem rohen Socket auf und **kappt** sie.

    Nur so entsteht wirklich das, was ein Klick auf das Kreuz erzeugt: eine
    Verbindung, die mitten im Leib endet. Ein sauber gesendeter, aber zu kurzer
    Leib wäre etwas anderes – den würde der Server als Formatfehler abweisen,
    und der Test bewiese nichts über den Abbruch.
    """
    u = urllib.parse.urlparse(BASE)
    grenze = "----aether" + uuid.uuid4().hex
    kopf = (f'--{grenze}\r\nContent-Disposition: form-data; name="folder_id"\r\n\r\n{folder_id}\r\n'
            f'--{grenze}\r\nContent-Disposition: form-data; name="file"; filename="{name}"\r\n'
            f'Content-Type: application/octet-stream\r\n\r\n').encode()
    schwanz = f"\r\n--{grenze}--\r\n".encode()
    laenge = len(kopf) + gesamt + len(schwanz)

    s = socket.create_connection((u.hostname, u.port or 80), timeout=10)
    anfrage = (
        f"POST /api/v1/files HTTP/1.1\r\n"
        f"Host: {u.hostname}:{u.port or 80}\r\n"
        f"Content-Type: multipart/form-data; boundary={grenze}\r\n"
        f"Content-Length: {laenge}\r\n"
        f"Cookie: {keksleiste(jar)}\r\n"
        f"X-CSRF-Token: {csrf['X-CSRF-Token']}\r\n"
        f"Connection: close\r\n\r\n"
    ).encode()
    s.sendall(anfrage + kopf)
    block = b"x" * 65536
    gesendet = 0
    while gesendet < nach_bytes:
        n = min(len(block), nach_bytes - gesendet)
        s.sendall(block[:n])
        gesendet += n
    # Hart schließen – kein sauberes Ende, genau wie ein fallengelassenes XHR.
    s.close()
    return gesendet


print(f"Upload-Abbruch-Selbsttest gegen {BASE}\n")

sess, jar, CSRF = anmelden()

ordner_name = "abbruch-test-" + uuid.uuid4().hex[:6]
st, ordner = ruf(sess, "POST", "/api/v1/folders", {"name": ordner_name}, CSRF)
assert st in (200, 201), f"Ordner anlegen fehlgeschlagen: {st}"
ORDNER = ordner["id"]


def aufraeumen():
    ruf(sess, "DELETE", f"/api/v1/folders/{ORDNER}", None, CSRF)
    # Auch aus dem Papierkorb, damit der Lauf nichts hinterlässt.
    st, papier = ruf(sess, "GET", "/api/v1/trash", None, CSRF)
    for e in (papier or []):
        if e.get("name", "").startswith("abbruch-test-") or e.get("name", "") in ABGELEGT:
            ruf(sess, "DELETE", f"/api/v1/trash/{e['id']}", None, CSRF)


ABGELEGT = set()

try:
    # ── 1. Neue Datei, mittendrin abgebrochen ────────────────────────────────
    print("Abgebrochener Upload einer neuen Datei")
    name = "abbruch-neu.bin"
    ABGELEGT.add(name)
    gesendet = abgebrochen_hochladen(jar, CSRF, name, ORDNER, 40 * 1024 * 1024, 4 * 1024 * 1024)
    check("Verbindung wurde mitten im Leib gekappt", gesendet == 4 * 1024 * 1024,
          f"gesendet={gesendet}")

    # Dem Server einen Moment lassen, den Abbruch zu bemerken und aufzuräumen.
    time.sleep(1.5)

    st, liste = ruf(sess, "GET", f"/api/v1/files?folder_id={ORDNER}", None, CSRF)
    namen = [f["name"] for f in (liste or {}).get("files", [])]
    check("keine Datei in der Datenbank", name not in namen, f"gefunden: {namen}")

    st, papier = ruf(sess, "GET", "/api/v1/trash", None, CSRF)
    check("auch nichts im Papierkorb", all(e["name"] != name for e in (papier or [])))

    # ── 2. Vorhandene Datei überschreiben und dabei abbrechen ────────────────
    print("\nAbbruch beim Überschreiben – die alte Fassung muss bleiben")
    inhalt = b"Die urspruengliche Fassung. " * 500
    pruefsum = hashlib.sha256(inhalt).hexdigest()
    name2 = "abbruch-vorhanden.bin"
    ABGELEGT.add(name2)
    datei = hochladen(sess, CSRF, name2, inhalt, ORDNER)
    check("Ausgangsdatei liegt", datei["size"] == len(inhalt), str(datei["size"]))

    # Versionierung an: Das ist der Weg, auf dem der Server die alte Fassung
    # kurzzeitig zur Seite schiebt – dort wäre ein Verlust am teuersten.
    st, _ = ruf(sess, "PATCH", f"/api/v1/files/{datei['id']}/versioning",
                {"enabled": True}, CSRF)
    check("Versionierung eingeschaltet", st == 200, f"status={st}")

    abgebrochen_hochladen(jar, CSRF, name2, ORDNER, 40 * 1024 * 1024, 4 * 1024 * 1024)
    time.sleep(1.5)

    st, liste = ruf(sess, "GET", f"/api/v1/files?folder_id={ORDNER}", None, CSRF)
    treffer = [f for f in (liste or {}).get("files", []) if f["name"] == name2]
    check("die Datei gibt es noch genau einmal", len(treffer) == 1, str(len(treffer)))
    if treffer:
        check("ihre Größe ist unverändert", treffer[0]["size"] == len(inhalt),
              f"{treffer[0]['size']} statt {len(inhalt)}")
        check("ihre Prüfsumme ist unverändert", treffer[0].get("checksum") == pruefsum,
              str(treffer[0].get("checksum"))[:16])

        # Und der Inhalt lässt sich wirklich noch holen – die Zeile in der
        # Datenbank nützt nichts, wenn die Bytes weggewandert sind.
        req = urllib.request.Request(BASE + f"/api/v1/files/{treffer[0]['id']}/download")
        with sess.open(req) as r:
            geladen = r.read()
        check("der Inhalt kommt unversehrt zurück", geladen == inhalt,
              f"{len(geladen)} statt {len(inhalt)} Bytes")

    # ── 3. Der Zähler stimmt: kein Platz für Gespenster ──────────────────────
    print("\nVerbrauch")
    st, me = ruf(sess, "GET", "/api/v1/auth/me", None, CSRF)
    check("Verbrauch abrufbar", st == 200 and me is not None, f"status={st}")
    check("Verbrauch ist eine Zahl", isinstance((me or {}).get("user", {}).get("used_bytes"), int),
          str((me or {}).get("user", {}).get("used_bytes")))
    # **/auth/me erneuert das CSRF-Token.** Wer das übersieht, bekommt beim
    # nächsten schreibenden Ruf ein 403 und sucht es an der falschen Stelle –
    # genau darauf bin ich beim Schreiben dieses Tests hereingefallen. Die
    # Weboberfläche zieht den neuen Wert nach (hooks/useAuth.ts), ein Skript
    # muss es auch.
    if me and me.get("csrf_token"):
        CSRF["X-CSRF-Token"] = me["csrf_token"]

    # ── 4. Verwaiste Reste? Der Rescan weiß es ───────────────────────────────
    print("\nRescan: liegen Bytes ohne Eintrag herum?")
    st, bericht = ruf(sess, "POST", "/api/v1/admin/rescan", {"apply": False}, CSRF)
    if st == 200 and bericht is not None:
        neu = bericht.get("new", 0)
        proben = [p.get("path", "") for p in (bericht.get("samples") or [])
                  if p.get("kind") == "new"]
        uebrig = [p for p in proben if "abbruch-" in p]
        check("keine herrenlosen Bytes vom Abbruch", not uebrig, str(uebrig))
        check("Rescan lief durch", isinstance(neu, int), str(neu))
    else:
        check("Rescan erreichbar", False, f"status={st}")

finally:
    aufraeumen()

print()
if fail:
    print(f"\033[31m{len(fail)} fehlgeschlagen\033[0m: " + ", ".join(fail))
print(f"{ok_count} bestanden, {len(fail)} fehlgeschlagen")
sys.exit(1 if fail else 0)