Aether/scripts/sharing-selbsttest.py

#!/usr/bin/env python3
"""Selbsttest der öffentlichen Freigabelinks.

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

Freigabelinks sind der **einzige** Teil von Aether, den man ohne Anmeldung
erreicht. Ein Fehler darin ist deshalb kein Anzeigefehler, sondern eine
offene Tür – und genau deshalb gibt es diesen Test.

Er legt einen Zweitnutzer, einen Ordner und Testdateien an und räumt am Ende
alles wieder weg. Vorhandene Dateien werden nicht angefasst.

**Nach einem Lauf ist die eigene IP 15 Minuten gesperrt** – für
passwortgeschützte Freigabelinks, nicht fürs Anmelden. Das ist der Preis der
letzten Prüfung, die genau diese Sperre nachweist. Ein zweiter Lauf innerhalb
der Frist bricht deshalb früh ab und sagt es.
"""
import base64
import http.cookiejar
import json
import sys
import urllib.error
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 = []
hinweise = []


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 hinweis(text):
    hinweise.append(text)
    print(f"  \033[33m!\033[0m    {text}")


def sitzung():
    """Eigener Cookie-Topf je Nutzer – sonst überschreibt die zweite
    Anmeldung die erste, und der Test prüft am Ende sich selbst."""
    jar = http.cookiejar.CookieJar()
    return urllib.request.build_opener(urllib.request.HTTPCookieProcessor(jar))


def ruf(opener, method, path, body=None, headers=None, roh=False):
    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 or urllib.request.build_opener()).open(req) as r:
            payload = r.read()
            if roh:
                return r.status, payload
            text = payload.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(user, passwort):
    o = sitzung()
    st, d = ruf(o, "POST", "/api/v1/auth/login", {"username": user, "password": passwort})
    assert st == 200, f"Anmeldung {user} fehlgeschlagen: {st}"
    return o, {"X-CSRF-Token": d["csrf_token"]}


def hochladen(opener, csrf, name, inhalt: bytes, folder_id=None):
    """Multipart von Hand – der Upload nimmt kein JSON."""
    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"]


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

admin, ADMIN_CSRF = anmelden(USER, PASSWORT)

# ── Zweitnutzer: nur mit ihm lässt sich prüfen, dass fremde Dateien fremd
#    bleiben. Ein Test, der nur das eigene Konto kennt, kann das nicht.
fremd_name = "selbsttest-fremd-" + uuid.uuid4().hex[:6]
fremd_pw = "Fremd!" + uuid.uuid4().hex[:12]
st, fremd = ruf(admin, "POST", "/api/v1/admin/users",
                {"username": fremd_name, "display_name": "Selbsttest Fremd",
                 "password": fremd_pw, "is_admin": False, "quota_bytes": 0}, ADMIN_CSRF)
assert st in (200, 201), f"Zweitnutzer anlegen fehlgeschlagen: {st}"
FREMD_ID = fremd["id"]

eigen = None
fremd_datei = None
links = []


def aufraeumen():
    for lid in links:
        ruf(admin, "DELETE", f"/api/v1/shares/{lid}", None, ADMIN_CSRF)
    if eigen:
        ruf(admin, "DELETE", f"/api/v1/files/{eigen['id']}", None, ADMIN_CSRF)
    # Löschen heißt Papierkorb – ein Test, der nur so weit aufräumt, hinterlässt
    # bei jedem Lauf eine Leiche mehr. Also auch dort hinterher.
    _, muell = ruf(admin, "GET", "/api/v1/trash")
    for f in muell or []:
        if f.get("name", "").startswith("selbsttest-"):
            ruf(admin, "DELETE", f"/api/v1/trash/{f['id']}", None, ADMIN_CSRF)
    ruf(admin, "DELETE", f"/api/v1/admin/users/{FREMD_ID}", None, ADMIN_CSRF)


try:
    INHALT = b"streng geheim " + uuid.uuid4().hex.encode()
    eigen = hochladen(admin, ADMIN_CSRF, "selbsttest-freigabe.txt", INHALT)

    fremd_op, FREMD_CSRF = anmelden(fremd_name, fremd_pw)
    fremd_datei = hochladen(fremd_op, FREMD_CSRF, "selbsttest-fremd.txt", b"gehoert mir")

    print("── Anlegen: nur eigene Dateien, nur angemeldet ──")
    st, _ = ruf(None, "POST", f"/api/v1/files/{eigen['id']}/share", {})
    check("ohne Anmeldung: kein Link", st in (401, 403), st)

    # Der Kern: Eine fremde Datei darf sich nicht freigeben lassen – sonst
    # macht ein beliebiger Nutzer fremde Daten öffentlich.
    st, _ = ruf(fremd_op, "POST", f"/api/v1/files/{eigen['id']}/share", {}, FREMD_CSRF)
    check("fremde Datei freigeben: 404", st == 404, st)
    st, _ = ruf(admin, "POST", f"/api/v1/files/{fremd_datei['id']}/share", {}, ADMIN_CSRF)
    check("auch als Admin nicht: 404", st == 404, st)

    st, link = ruf(admin, "POST", f"/api/v1/files/{eigen['id']}/share", {}, ADMIN_CSRF)
    check("eigene Datei freigeben (201)", st == 201, st)
    links.append(link["id"])
    TOKEN = link["token"]

    # 24 zufällige Bytes, base64url = 32 Zeichen. Raten ist damit aussichtslos;
    # ein kurzer Token wäre die stille Version einer offenen Tür.
    check("Token ist lang genug (>= 32 Zeichen)", len(TOKEN) >= 32, len(TOKEN))

    print("\n── Öffentlich abrufen ──")
    st, info = ruf(None, "GET", f"/share/{TOKEN}")
    check("Info ohne Anmeldung abrufbar", st == 200, st)
    check("nennt Namen und Größe",
          info and info.get("file_name") == "selbsttest-freigabe.txt"
          and info.get("file_size") == len(INHALT), info)
    # Was der Server öffentlich preisgibt, muss man einmal ausdrücklich prüfen:
    # Ein durchgereichtes Datenbankfeld zu viel, und der Hash steht im Netz.
    flach = json.dumps(info).lower()
    check("kein Passwort-Hash in der Antwort",
          "password_hash" not in flach and "$argon2" not in flach, flach[:200])
    check("keine Angabe zum Besitzer", "owner_id" not in flach, flach[:200])

    st, _ = ruf(None, "GET", "/share/" + "x" * 32)
    check("erfundener Token: 404", st == 404, st)

    st, daten = ruf(None, "POST", f"/share/{TOKEN}/download", {}, roh=True)
    check("Herunterladen ohne Passwort (200)", st == 200, st)
    check("Inhalt stimmt byte-genau", daten == INHALT, daten[:40])

    st, info = ruf(None, "GET", f"/share/{TOKEN}")
    check("Zähler steht auf 1", info and info["download_count"] == 1,
          info["download_count"] if info else None)

    print("\n── Mit Passwort ──")
    st, geschuetzt = ruf(admin, "POST", f"/api/v1/files/{eigen['id']}/share",
                         {"password": "geheim123"}, ADMIN_CSRF)
    links.append(geschuetzt["id"])
    GTOKEN = geschuetzt["token"]
    check("Link mit Passwort angelegt", st == 201 and geschuetzt["has_password"] is True, st)

    st, _ = ruf(None, "POST", f"/share/{GTOKEN}/download", {})
    if st == 429:
        # Die Sperre aus einem Lauf vor weniger als 15 Minuten steht noch.
        # Das ist richtiges Verhalten des Servers, aber der Rest dieses
        # Abschnitts kann so nichts mehr aussagen.
        hinweis("IP ist noch gesperrt (Lauf vor weniger als 15 Minuten) – "
                "Passwortprüfungen übersprungen, bitte später erneut laufen lassen")
        raise SystemExit(1 if fail else 0)
    check("ohne Passwort: 401", st == 401, st)
    st, _ = ruf(None, "POST", f"/share/{GTOKEN}/download", {"password": "falsch"})
    check("falsches Passwort: 401", st == 401, st)
    st, daten = ruf(None, "POST", f"/share/{GTOKEN}/download", {"password": "geheim123"}, roh=True)
    check("richtiges Passwort: Inhalt", st == 200 and daten == INHALT, st)

    print("\n── Widerrufen ──")
    st, _ = ruf(fremd_op, "DELETE", f"/api/v1/shares/{link['id']}", None, FREMD_CSRF)
    check("fremder Link lässt sich nicht löschen", st == 404, st)

    st, _ = ruf(admin, "DELETE", f"/api/v1/shares/{link['id']}", None, ADMIN_CSRF)
    check("eigener Link gelöscht", st in (200, 204), st)
    links.remove(link["id"])
    st, _ = ruf(None, "GET", f"/share/{TOKEN}")
    check("danach ist der Link tot (404)", st == 404, st)
    st, _ = ruf(None, "POST", f"/share/{TOKEN}/download", {})
    check("und das Herunterladen auch", st == 404, st)

    print("\n── Gelöschte Datei ──")
    st, fluechtig = ruf(admin, "POST", f"/api/v1/files/{eigen['id']}/share", {}, ADMIN_CSRF)
    links.append(fluechtig["id"])
    ruf(admin, "DELETE", f"/api/v1/files/{eigen['id']}", None, ADMIN_CSRF)
    # In den Papierkorb geworfen heißt: für die Öffentlichkeit weg. Ein Link,
    # der weiterläuft, während die Datei im Papierkorb liegt, wäre eine
    # Freigabe, von der der Besitzer nichts mehr weiß.
    st, _ = ruf(None, "POST", f"/share/{fluechtig['token']}/download", {})
    check("Link auf gelöschte Datei liefert nichts", st in (404, 410), st)

    # Zum Schluss, mit Absicht: Diese Probe sperrt die eigene IP für 15
    # Minuten. Stünde sie weiter oben, fielen alle folgenden Passwortprüfungen
    # in die selbst gestellte Falle.
    print("\n── Durchprobieren von Passwörtern ──")
    codes = [ruf(None, "POST", f"/share/{GTOKEN}/download", {"password": f"f{i}"})[0]
             for i in range(15)]
    check("Sperre greift nach wiederholten Fehlversuchen (429)", 429 in codes,
          f"letzte Antworten: {codes[-5:]}")
    check("und sie greift erst nach mehreren Versuchen", codes[0] == 401, codes[:3])

finally:
    aufraeumen()

print(f"\n{ok_count} bestanden, {len(fail)} fehlgeschlagen")
for h in hinweise:
    print(f"Hinweis: {h}")
if fail:
    print("Fehlgeschlagen: " + ", ".join(fail))
sys.exit(1 if fail else 0)