marcandClaude Opus 5
6fab2388ca
Beim Lesen von internal/sharing fiel eine ungleiche Zusage auf: Der angemeldete Download prüft seit dem 12.09. die SHA-256-Summe beim Ausliefern und lässt die Verbindung fallen, wenn sie nicht stimmt. Der öffentliche Freigabelink lief durch ein blankes io.Copy – ungeprüft. Ausgerechnet der Weg nach draussen. Wer die Datei angemeldet holt, ist der Eigentümer und merkt, wenn etwas kaputt ist; wer einen Link bekommt, merkt es nicht. In CLAUDE.md steht das unter „nicht verhandelbar": beim Upload berechnen, beim Download verifizieren. Die Regel galt an einer von zwei Stellen. Die Mechanik liegt jetzt in files/verified.go und wird von beiden gerufen. Nachgewiesen mit gekipptem Bit auf der Platte: Über den Freigabelink kommt die Datei ebenfalls nicht mehr vollständig an. Gruppenordner: Create übergab user.ID statt des aufgelösten Prinzipals – eine Datei im Gruppenordner liess sich nicht verlinken, auch nicht vom Mitglied. Marcs Entscheidung: Das soll gehen, der Empfänger sieht ohnehin nur die Datei, nie den Ordner darum. Verlangt wird Schreibrecht: Einen Link zu erzeugen heisst, die Datei öffentlich zu machen; wer sie nur geliehen lesen darf, soll sie nicht weiterreichen können (403). expires_in_hours hatte keine Obergrenze. time.Duration rechnet in Nanosekunden; ab etwa 2,6 Mio Stunden läuft int64 über, die Dauer wird negativ, und der Link ist abgelaufen in dem Moment, in dem er entsteht – ohne Fehlermeldung. Jetzt zehn Jahre, darüber 400. Dazu X-Content-Type-Options: nosniff, das der angemeldete Weg längst setzt. Sechs Go-Tests für internal/sharing, unter Mutation nachgewiesen: Eigentümerprüfung entfernt, Ablaufprüfung entfernt und Token verkürzt machen jeweils genau den zugehörigen Test rot. Selbsttests erweitert: sharing 24 → 31 Prüfungen, pruefsumme 14 → 16. Geprüft: go build/vet sauber, alle Go-Pakete grün, 241 Prüfungen der sieben Selbsttest-Suiten grün, 48 Vitest-Tests. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
332 lines
15 KiB
Python
332 lines
15 KiB
Python
#!/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 mit_kopfzeilen(method, path, body=None):
|
||
"""Wie `ruf`, gibt aber die Antwortkopfzeilen zurueck. Gebraucht fuer
|
||
nosniff – das steht nirgends im Rumpf."""
|
||
data = json.dumps(body).encode() if body is not None else b""
|
||
req = urllib.request.Request(BASE + path, data=data, method=method,
|
||
headers={"Content-Type": "application/json"})
|
||
try:
|
||
with urllib.request.build_opener().open(req) as r:
|
||
r.read()
|
||
return r.status, dict(r.headers)
|
||
except urllib.error.HTTPError as e:
|
||
return e.code, dict(e.headers)
|
||
|
||
|
||
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)
|
||
# **Achtung:** `/auth/me` dreht den CSRF-Token weiter (siehe TESTING.md). Wer
|
||
# danach den alten weiterbenutzt, bekommt 403 – also den neuen uebernehmen.
|
||
_, ich = ruf(admin, "GET", "/api/v1/auth/me")
|
||
ADMIN_ID = ich["user"]["id"]
|
||
ADMIN_CSRF = {"X-CSRF-Token": ich["csrf_token"]}
|
||
|
||
# ── 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 = []
|
||
gruppen = []
|
||
dateien = []
|
||
|
||
|
||
def aufraeumen():
|
||
for lid in links:
|
||
ruf(admin, "DELETE", f"/api/v1/shares/{lid}", None, ADMIN_CSRF)
|
||
for did in dateien:
|
||
ruf(admin, "DELETE", f"/api/v1/files/{did}", 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)
|
||
for gid in gruppen:
|
||
ruf(admin, "DELETE", f"/api/v1/admin/groups/{gid}", 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)
|
||
|
||
# ── Seit dem 25.09.2026 ────────────────────────────────────────────────
|
||
print("\n── Gruppenordner, Grenzen, Kopfzeilen ──")
|
||
|
||
# **Gruppendatei teilbar.** Vorher übergab Create `user.ID` statt des
|
||
# aufgelösten Prinzipals – eine Datei im Gruppenordner liess sich nicht
|
||
# verlinken, auch nicht vom Mitglied. Die Antwort lautete „file not found".
|
||
gname = "selbsttest-gruppe-" + uuid.uuid4().hex[:6]
|
||
st, gruppe = ruf(admin, "POST", "/api/v1/admin/groups",
|
||
{"name": gname, "description": "Selbsttest"}, ADMIN_CSRF)
|
||
if st in (200, 201):
|
||
gruppen.append(gruppe["id"])
|
||
ruf(admin, "POST", f"/api/v1/admin/groups/{gruppe['id']}/members",
|
||
{"user_id": ADMIN_ID}, ADMIN_CSRF)
|
||
# Der Wurzelordner der Gruppe taucht in der normalen Ordnerliste auf.
|
||
_, wurzeln = ruf(admin, "GET", "/api/v1/files")
|
||
gordner = next((f for f in (wurzeln or {}).get("folders", [])
|
||
if f.get("group_id")), None)
|
||
check("Gruppenordner ist da", gordner is not None)
|
||
if gordner:
|
||
gdatei = hochladen(admin, ADMIN_CSRF, "selbsttest-gruppe.txt",
|
||
b"in der gruppe", gordner["id"])
|
||
dateien.append(gdatei["id"])
|
||
st, glink = ruf(admin, "POST", f"/api/v1/files/{gdatei['id']}/share",
|
||
{}, ADMIN_CSRF)
|
||
check("Datei im Gruppenordner laesst sich verlinken", st == 201, st)
|
||
if st == 201:
|
||
links.append(glink["id"])
|
||
st, _ = ruf(None, "GET", f"/share/{glink['token']}")
|
||
check("und der Link funktioniert oeffentlich", st == 200, st)
|
||
|
||
# Frische Datei: Die weiter oben ist inzwischen geloescht – die Abschnitte
|
||
# davor pruefen genau das.
|
||
frisch = hochladen(admin, ADMIN_CSRF, "selbsttest-frist.txt", b"frist")
|
||
dateien.append(frisch["id"])
|
||
|
||
# **Obergrenze fuer den Ablauf.** Ohne sie laeuft time.Duration ueber, die
|
||
# Frist wird negativ, und der Link ist tot in dem Moment, in dem er
|
||
# entsteht – ohne Fehlermeldung.
|
||
st, _ = ruf(admin, "POST", f"/api/v1/files/{frisch['id']}/share",
|
||
{"expires_in_hours": 9_000_000_000}, ADMIN_CSRF)
|
||
check("unsinnige Ablauffrist wird abgelehnt (400)", st == 400, st)
|
||
|
||
st, kurz = ruf(admin, "POST", f"/api/v1/files/{frisch['id']}/share",
|
||
{"expires_in_hours": 24}, ADMIN_CSRF)
|
||
check("vernuenftige Ablauffrist geht durch", st == 201, st)
|
||
if st == 201:
|
||
links.append(kurz["id"])
|
||
st, kopf = mit_kopfzeilen("POST", f"/share/{kurz['token']}/download")
|
||
check("oeffentlicher Download setzt nosniff",
|
||
kopf.get("X-Content-Type-Options") == "nosniff", dict(kopf))
|
||
|
||
# **Nur-Lese-Freigabe darf nicht weiterverteilen.** Wer eine Datei geliehen
|
||
# bekommt, soll sie nicht oeffentlich machen koennen.
|
||
st, _ = ruf(admin, "POST", f"/api/v1/files/{frisch['id']}/acl",
|
||
{"subject_id": FREMD_ID, "subject_type": "user",
|
||
"permission": "read"}, ADMIN_CSRF)
|
||
if st in (200, 201):
|
||
fremd_sitzung, FREMD_CSRF = anmelden(fremd_name, fremd_pw)
|
||
st, _ = ruf(fremd_sitzung, "POST", f"/api/v1/files/{frisch['id']}/share",
|
||
{}, FREMD_CSRF)
|
||
check("nur geliehen gelesen -> kein eigener Freigabelink (403)",
|
||
st == 403, 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)
|