security(sharing): Selbsttest fuer Freigabelinks, drei Befunde behoben
Freigabelinks sind der einzige Teil von Aether, den man ohne Anmeldung erreicht -- und hatten als einziger Bereich keinen einzigen Test. Der neue scripts/sharing-selbsttest.py geht sie mit 24 Pruefungen ab, mit einem eigens angelegten Zweitnutzer, ohne den sich "fremde Dateien bleiben fremd" gar nicht pruefen laesst. Beim ersten Lauf sofort drei Befunde: 1. Der Download-Zaehler blieb auf 0. Die Zaehlung lief in einer Goroutine mit dem Kontext der Anfrage -- der wird abgebrochen, sobald die Antwort draussen ist, und die Zaehlung verlor das Rennen fast immer. Jetzt context.WithoutCancel mit eigener Frist. 2. owner_id, file_id und die Link-Kennung standen in der oeffentlichen Antwort. Das ganze Datenbankobjekt durchzureichen war bequem, gab aber die Kennung des Besitzers an jeden heraus, der den Link hat: Zwei Links desselben Besitzers liessen sich einander zuordnen, ohne dass irgendwer angemeldet ist. Jetzt nur noch, was die oeffentliche Seite braucht. 3. Passwoerter liessen sich unbegrenzt durchprobieren. Der oeffentliche Freigabepfad lag ausserhalb jeder Ratenbegrenzung, und weil jeder Versuch Argon2 rechnet, war das zugleich ein billiger Weg, den Server zu beschaeftigen. Jetzt dieselbe Fehlversuchssperre wie bei /dav und /caldav: zehn Fehlschlaege in 15 Minuten, dann 15 Minuten Pause. Die Sperre hat eine Folge fuers Testen selbst: Nach einem Lauf ist die eigene IP 15 Minuten fuer passwortgeschuetzte Links gesperrt. Deshalb steht die Probe am Ende, und der Test erkennt zu Beginn, ob eine Sperre aus einem frueheren Lauf noch steht, statt reihenweise Fehlschlaege zu melden, die keine sind. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
marcandClaude Opus 5 committed
1 parent
1e1f35955d
commit
ec8f162fd5
5 files changed
+333
-11
No files matched your search
@@ -0,0 +1,237 @@
|
||||
#!/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"
|
||||
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)
|
||||
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)
|
||||
Reference in new issue
Block a user