#!/usr/bin/env python3 """Selbsttest: Wird eine beschädigte Datei beim Herunterladen erkannt? python3 scripts/pruefsumme-selbsttest.py [BASIS-URL] [DATENPFAD] [BENUTZER] [PASSWORT] Aether speichert zu jeder Datei eine SHA-256-Summe und rechnet sie beim Ausliefern mit. Bis zum 12.09.2026 wurde das Ergebnis **weggeworfen**: Ein gekipptes Bit auf der Platte wurde erkannt und verschwiegen, und der Empfänger bekam eine Datei, die heil aussah und keine war. Der Test kippt deshalb wirklich ein Bit – im Blob auf der Platte, nicht im Programm – und prüft, was danach herauskommt. Geprüft wird: * Eine heile Datei kommt unverändert durch (sonst bewiese der Rest nichts). * Eine beschädigte Datei kommt **nicht vollständig** an. * Der Server sagt es im Log, mit Datei-ID, erwarteter und gelesener Summe. * Groß und klein: über und unter den 32 KiB, die bis zur Prüfung zurückgehalten werden – beim kleinen Fall wird gar nichts herausgegeben. Der Test muss auf dem Rechner laufen, auf dem die Daten liegen, und braucht Docker-Zugriff für den Logeintrag. """ import hashlib import http.cookiejar import json import os import subprocess import sys import urllib.error import urllib.request import uuid BASE = (sys.argv[1] if len(sys.argv) > 1 else "http://localhost:8090").rstrip("/") DATEN = sys.argv[2] if len(sys.argv) > 2 else os.path.expanduser("~/aether-data") USER = sys.argv[3] if len(sys.argv) > 3 else "admin" PASSWORT = sys.argv[4] if len(sys.argv) > 4 else "admin123" CONTAINER = os.environ.get("AETHER_CONTAINER", "aether-app") 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}") jar = http.cookiejar.CookieJar() OPENER = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(jar)) def api(method, path, body=None, headers=None): hdrs = dict(headers or {}) data = json.dumps(body).encode() if body is not None else None if body is not None: 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 hochladen(name, inhalt: bytes, csrf): grenze = "----aether" + uuid.uuid4().hex 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 = 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 herunterladen(file_id): """(vollstaendig, bytes|None, fehler)""" req = urllib.request.Request(BASE + f"/api/v1/files/{file_id}/download") try: with OPENER.open(req) as r: erwartet = int(r.headers.get("Content-Length", "-1")) daten = r.read() return (erwartet < 0 or len(daten) == erwartet), daten, "" except Exception as e: # Abgebrochene Antwort: genau das ist der gewuenschte Ausgang. return False, None, type(e).__name__ + ": " + str(e)[:80] def oeffentlich_holen(token): """Dasselbe ueber den Freigabelink – ohne Anmeldung. Bis zum 25.09.2026 lief hier ein blankes io.Copy: Wer die Datei ueber einen Link bekam, bekam sie ungeprueft. Ausgerechnet auf dem Weg, auf dem sie das Haus verlaesst. """ req = urllib.request.Request(BASE + f"/share/{token}/download", data=b"{}", method="POST", headers={"Content-Type": "application/json"}) try: with urllib.request.build_opener().open(req) as r: erwartet = int(r.headers.get("Content-Length", "-1")) daten = r.read() return (erwartet < 0 or len(daten) == erwartet), daten, "" except Exception as e: return False, None, type(e).__name__ + ": " + str(e)[:80] def blobpfad(name): """Wo der Blob liegt. **Nicht** aus der Antwort: Der interne Speicherpfad steht dort bewusst nicht drin, er geht einen Client nichts an. Für eine Datei im Wurzelverzeichnis ist er aber vorhersagbar – siehe BuildStorageKey. """ return os.path.join(DATEN, "users", USER, "home", name) def logzeilen(seit): try: aus = subprocess.run(["docker", "logs", "--since", seit, CONTAINER], capture_output=True, text=True, timeout=20) return aus.stdout + aus.stderr except Exception: return "" print(f"Pruefsummen-Selbsttest gegen {BASE}\n") st, login = api("POST", "/api/v1/auth/login", {"username": USER, "password": PASSWORT}) assert st == 200, f"Anmeldung fehlgeschlagen: {st}" CSRF = {"X-CSRF-Token": login["csrf_token"]} angelegt = [] geteilt = [] def aufraeumen(): for lid in geteilt: api("DELETE", f"/api/v1/shares/{lid}", None, CSRF) for fid in angelegt: api("DELETE", f"/api/v1/files/{fid}", None, CSRF) st, papier = api("GET", "/api/v1/trash", None, CSRF) for e in (papier or []): if e.get("name", "").startswith("pruefsumme-"): api("DELETE", f"/api/v1/trash/{e['id']}", None, CSRF) try: for bezeichnung, groesse in [("gross", 200 * 1024), ("klein", 900)]: print(f"── {bezeichnung}e Datei ({groesse} Bytes) ──") name = f"pruefsumme-{bezeichnung}-{uuid.uuid4().hex[:6]}.bin" inhalt = bytes((i * 7 + 13) % 256 for i in range(groesse)) datei = hochladen(name, inhalt, CSRF) angelegt.append(datei["id"]) check("hochgeladen", datei["size"] == groesse, datei["size"]) check("Pruefsumme stimmt mit der eigenen ueberein", datei["checksum"] == hashlib.sha256(inhalt).hexdigest(), datei["checksum"][:16]) # 1. Heil: Das muss durchgehen, sonst beweist der Rest nichts. voll, daten, fehler = herunterladen(datei["id"]) check("heile Datei kommt vollstaendig an", voll and daten == inhalt, fehler or f"{len(daten or b'')} von {groesse}") # 2. Ein Bit kippen – auf der Platte, nicht im Programm. pfad = blobpfad(name) if not os.path.isfile(pfad): check(f"Blob gefunden ({pfad})", False, "nicht da – falscher DATENPFAD?") continue with open(pfad, "r+b") as f: # In der Mitte, damit auch der zurueckgehaltene Schluss heil bleibt: # Der Fehler darf nicht schon an der Laenge auffallen. f.seek(groesse // 2) b = f.read(1) f.seek(groesse // 2) f.write(bytes([b[0] ^ 0xFF])) check("ein Bit auf der Platte gekippt", True) # 3. Jetzt darf sie nicht mehr heil herauskommen. marke = "1m" voll, daten, fehler = herunterladen(datei["id"]) check("beschaedigte Datei kommt NICHT vollstaendig an", not voll, f"{len(daten or b'')} von {groesse} Bytes – heil ausgeliefert!") if daten is not None: check("und was ankam, ist kuerzer als angekuendigt", len(daten) < groesse, f"{len(daten)} von {groesse}") # 5. Und jetzt derselbe Weg nach draussen: ueber einen Freigabelink. # Die Datei ist dieselbe und immer noch kaputt – die Zusage muss # hier genauso gelten. st, link = api("POST", f"/api/v1/files/{datei['id']}/share", {}, CSRF) if st == 201 and link: geteilt.append(link["id"]) voll, daten, fehler = oeffentlich_holen(link["token"]) check("auch ueber den Freigabelink kommt sie NICHT vollstaendig an", not voll, f"{len(daten or b'')} von {groesse} Bytes – heil ausgeliefert!") else: check("Freigabelink angelegt", False, st) # 4. Der Server muss es gesagt haben. log = logzeilen(marke) if log: check("Server meldet die Beschaedigung im Log", "Pruefsumme stimmt nicht" in log, log[-200:] if log else "") check("mit der Datei-ID", datei["id"] in log, "") else: print(" \033[33m!\033[0m Log nicht lesbar (docker logs) – Logpruefung uebersprungen") print() finally: aufraeumen() 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)