guyg1474andClaude Opus 5.5
0fb3521c2f
- Freunde (freunde.py, Tabelle friendships): suchen, anfragen, annehmen, ablehnen, entfernen; neue Seite "Freunde" im Menü - Vorratslisten und Kanban-Projekte immer privat: sichtbar für Ersteller und Mitglieder; teilen nur mit Freunden (Admins: mit allen); Kanban-Mitglieder (Tabelle kanban_project_members), Bearbeiter nur aus dem Projekt, ausgetretene Mitglieder verlieren ihre Zuteilung - Artikel: lesen alle; bearbeiten Autor, Mitarbeitende verknüpfter Kanban-Projekte und Admins; verknüpfen nur bearbeitbare Artikel (sonst ließen sich Rechte erschleichen); Board-Liste auf der Artikelseite zeigt nur sichtbare Boards - Administratoren sehen und bearbeiten alles - Reisen: Städte nur von Freunden (Admins: alle); "Reiserouten zeigen" zeichnet die Strecken der Kilometer-Rechnung als Bögen - Migration Schritt 6: bisher gemeinsame Listen und alle Projekte bekommen die vorhandenen Konten als Mitglieder, ohne Ersteller → erster Admin; erste Liste einer neuen Installation gehört dem ersten Konto - Tests: test_freunde.py, test_zugriff.py, bestehende auf Privat-Modell umgestellt Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
293 lines
12 KiB
Python
293 lines
12 KiB
Python
"""Konten, Anmeldung (Sitzungs-Cookie) und Benutzerverwaltung."""
|
||
import hashlib
|
||
import json
|
||
import re
|
||
import secrets
|
||
import time
|
||
from collections import defaultdict
|
||
|
||
from fastapi import APIRouter, Body, Depends, HTTPException, Request, Response
|
||
|
||
from .config import LINKS as CONFIG_LINKS
|
||
from .common import now_iso
|
||
from .db import get_meta, transaction
|
||
|
||
router = APIRouter()
|
||
|
||
SESSION_COOKIE = "p42_session"
|
||
SESSION_SECONDS = 30 * 24 * 60 * 60
|
||
USERNAME = re.compile(r"[A-Za-z0-9._-]{3,32}")
|
||
MIN_PASSWORD = 8
|
||
|
||
# ---------- Passwörter ----------
|
||
|
||
_SCRYPT = {"n": 2**14, "r": 8, "p": 1, "dklen": 32}
|
||
|
||
|
||
def hash_password(password: str) -> str:
|
||
salt = secrets.token_bytes(16)
|
||
digest = hashlib.scrypt(password.encode(), salt=salt, **_SCRYPT)
|
||
return f"scrypt${salt.hex()}${digest.hex()}"
|
||
|
||
|
||
def verify_password(password: str, stored: str) -> bool:
|
||
_, salt, digest = stored.split("$")
|
||
calc = hashlib.scrypt(password.encode(), salt=bytes.fromhex(salt), **_SCRYPT)
|
||
return secrets.compare_digest(calc.hex(), digest)
|
||
|
||
|
||
# Für unbekannte Benutzernamen trotzdem rechnen, damit die Antwortzeit nichts verrät
|
||
_DUMMY_HASH = hash_password(secrets.token_hex(8))
|
||
|
||
# ---------- Hilfsfunktionen ----------
|
||
|
||
|
||
def user_dict(row) -> dict:
|
||
return {
|
||
"id": row["id"],
|
||
"username": row["username"],
|
||
"displayName": row["display_name"],
|
||
"isAdmin": bool(row["is_admin"]),
|
||
"createdAt": row["created_at"],
|
||
}
|
||
|
||
|
||
def _hash_token(token: str) -> str:
|
||
return hashlib.sha256(token.encode()).hexdigest()
|
||
|
||
|
||
def _username(value) -> str:
|
||
v = value.strip() if isinstance(value, str) else ""
|
||
if not USERNAME.fullmatch(v):
|
||
raise HTTPException(400, "Benutzername: 3–32 Zeichen, nur Buchstaben, Zahlen, Punkt, Minus und Unterstrich.")
|
||
return v
|
||
|
||
|
||
def _display_name(value, fallback: str = "") -> str:
|
||
v = value.strip() if isinstance(value, str) else ""
|
||
v = v or fallback
|
||
if not v or len(v) > 60:
|
||
raise HTTPException(400, "Anzeigename fehlt oder ist zu lang (max. 60 Zeichen).")
|
||
return v
|
||
|
||
|
||
def _password(value) -> str:
|
||
if not isinstance(value, str) or len(value) < MIN_PASSWORD:
|
||
raise HTTPException(400, f"Das Passwort muss mindestens {MIN_PASSWORD} Zeichen lang sein.")
|
||
if len(value) > 200:
|
||
raise HTTPException(400, "Das Passwort ist zu lang.")
|
||
return value
|
||
|
||
|
||
def _start_session(response: Response, request: Request, user_id: int) -> None:
|
||
token = secrets.token_urlsafe(32)
|
||
with transaction() as conn:
|
||
conn.execute("DELETE FROM sessions WHERE expires_at < ?", (time.time(),))
|
||
conn.execute(
|
||
"INSERT INTO sessions (token_hash, user_id, expires_at) VALUES (?, ?, ?)",
|
||
(_hash_token(token), user_id, time.time() + SESSION_SECONDS),
|
||
)
|
||
https = request.headers.get("x-forwarded-proto", request.url.scheme) == "https"
|
||
response.set_cookie(
|
||
SESSION_COOKIE, token, max_age=SESSION_SECONDS, httponly=True, samesite="lax", secure=https, path="/",
|
||
)
|
||
|
||
|
||
def _create_user(conn, username: str, display_name: str, password: str, is_admin: bool) -> int:
|
||
if conn.execute("SELECT 1 FROM users WHERE username = ?", (username,)).fetchone():
|
||
raise HTTPException(400, "Diesen Benutzernamen gibt es schon.")
|
||
cur = conn.execute(
|
||
"INSERT INTO users (username, display_name, password_hash, is_admin, created_at) VALUES (?, ?, ?, ?, ?)",
|
||
(username, display_name, hash_password(password), int(is_admin), now_iso()),
|
||
)
|
||
return cur.lastrowid
|
||
|
||
|
||
# ---------- Abhängigkeiten für geschützte Routen ----------
|
||
|
||
|
||
def current_user(request: Request) -> dict:
|
||
token = request.cookies.get(SESSION_COOKIE)
|
||
if token:
|
||
with transaction() as conn:
|
||
row = conn.execute(
|
||
"""SELECT u.* FROM sessions s JOIN users u ON u.id = s.user_id
|
||
WHERE s.token_hash = ? AND s.expires_at > ?""",
|
||
(_hash_token(token), time.time()),
|
||
).fetchone()
|
||
if row:
|
||
return user_dict(row)
|
||
raise HTTPException(401, "Nicht angemeldet.")
|
||
|
||
|
||
def require_admin(user: dict = Depends(current_user)) -> dict:
|
||
if not user["isAdmin"]:
|
||
raise HTTPException(403, "Nur für Administratoren.")
|
||
return user
|
||
|
||
|
||
# ---------- Anmelde-Schutz: max. 5 Fehlversuche in 10 Minuten ----------
|
||
|
||
_failures: dict[tuple, list[float]] = defaultdict(list)
|
||
MAX_FAILURES, WINDOW = 5, 600
|
||
|
||
|
||
def _check_rate_limit(key: tuple) -> None:
|
||
now = time.time()
|
||
_failures[key] = [t for t in _failures[key] if now - t < WINDOW]
|
||
if len(_failures[key]) >= MAX_FAILURES:
|
||
raise HTTPException(429, "Zu viele Fehlversuche. Bitte in ein paar Minuten erneut versuchen.")
|
||
|
||
|
||
# ---------- Routen: Anmeldung ----------
|
||
|
||
|
||
@router.get("/api/auth/status")
|
||
def status(request: Request):
|
||
with transaction() as conn:
|
||
has_users = conn.execute("SELECT 1 FROM users LIMIT 1").fetchone() is not None
|
||
try:
|
||
user = current_user(request)
|
||
except HTTPException:
|
||
user = None
|
||
return {"setupRequired": not has_users, "user": user}
|
||
|
||
|
||
@router.post("/api/auth/setup")
|
||
def setup(request: Request, response: Response, data: dict = Body(...)):
|
||
"""Erstes Konto anlegen – wird Administrator und übernimmt die bisherigen Daten."""
|
||
with transaction() as conn:
|
||
if conn.execute("SELECT 1 FROM users LIMIT 1").fetchone():
|
||
raise HTTPException(403, "Es gibt bereits ein Konto.")
|
||
username = _username(data.get("username"))
|
||
display_name = _display_name(data.get("displayName"), username)
|
||
password = _password(data.get("password"))
|
||
with transaction() as conn:
|
||
if conn.execute("SELECT 1 FROM users LIMIT 1").fetchone():
|
||
raise HTTPException(403, "Es gibt bereits ein Konto.")
|
||
user_id = _create_user(conn, username, display_name, password, is_admin=True)
|
||
# Bisherige Projekte ohne Autor diesem Konto zuordnen
|
||
conn.execute("UPDATE projects SET created_by = ? WHERE created_by IS NULL", (user_id,))
|
||
conn.execute("UPDATE projects SET updated_by = ? WHERE updated_by IS NULL", (user_id,))
|
||
# Listen und Kanban-Projekte ohne Besitzer (z. B. die erste Liste "Zuhause") gehören ihm
|
||
conn.execute("UPDATE shopping_lists SET created_by = ? WHERE created_by IS NULL", (user_id,))
|
||
conn.execute("UPDATE kanban_projects SET created_by = ? WHERE created_by IS NULL", (user_id,))
|
||
# Bisherige Links (links.json bzw. config.json) übernehmen
|
||
legacy = get_meta(conn, "legacy_links")
|
||
links = json.loads(legacy) if legacy else CONFIG_LINKS
|
||
for key, url in (links or {}).items():
|
||
if url:
|
||
conn.execute("INSERT OR REPLACE INTO user_links VALUES (?, ?, ?)", (user_id, key, url))
|
||
conn.execute("DELETE FROM meta WHERE key = 'legacy_links'")
|
||
row = conn.execute("SELECT * FROM users WHERE id = ?", (user_id,)).fetchone()
|
||
_start_session(response, request, user_id)
|
||
return user_dict(row)
|
||
|
||
|
||
@router.post("/api/auth/login")
|
||
def login(request: Request, response: Response, data: dict = Body(...)):
|
||
username = str(data.get("username") or "").strip()
|
||
password = str(data.get("password") or "")
|
||
key = (request.client.host if request.client else "", username.lower())
|
||
_check_rate_limit(key)
|
||
with transaction() as conn:
|
||
row = conn.execute("SELECT * FROM users WHERE username = ?", (username,)).fetchone()
|
||
if not verify_password(password, row["password_hash"] if row else _DUMMY_HASH) or not row:
|
||
_failures[key].append(time.time())
|
||
raise HTTPException(401, "Benutzername oder Passwort falsch.")
|
||
_failures.pop(key, None)
|
||
_start_session(response, request, row["id"])
|
||
return user_dict(row)
|
||
|
||
|
||
@router.post("/api/auth/logout")
|
||
def logout(request: Request, response: Response):
|
||
token = request.cookies.get(SESSION_COOKIE)
|
||
if token:
|
||
with transaction() as conn:
|
||
conn.execute("DELETE FROM sessions WHERE token_hash = ?", (_hash_token(token),))
|
||
response.delete_cookie(SESSION_COOKIE, path="/")
|
||
return {"ok": True}
|
||
|
||
|
||
@router.patch("/api/auth/me")
|
||
def update_me(request: Request, data: dict = Body(...), user: dict = Depends(current_user)):
|
||
"""Eigenen Anzeigenamen oder eigenes Passwort ändern."""
|
||
with transaction() as conn:
|
||
if "displayName" in data:
|
||
conn.execute("UPDATE users SET display_name = ? WHERE id = ?", (_display_name(data["displayName"]), user["id"]))
|
||
if data.get("newPassword"):
|
||
row = conn.execute("SELECT password_hash FROM users WHERE id = ?", (user["id"],)).fetchone()
|
||
if not verify_password(str(data.get("currentPassword") or ""), row["password_hash"]):
|
||
raise HTTPException(400, "Das aktuelle Passwort stimmt nicht.")
|
||
conn.execute("UPDATE users SET password_hash = ? WHERE id = ?", (hash_password(_password(data["newPassword"])), user["id"]))
|
||
# Andere Sitzungen (z. B. auf anderen Geräten) beenden
|
||
current = _hash_token(request.cookies.get(SESSION_COOKIE, ""))
|
||
conn.execute("DELETE FROM sessions WHERE user_id = ? AND token_hash != ?", (user["id"], current))
|
||
row = conn.execute("SELECT * FROM users WHERE id = ?", (user["id"],)).fetchone()
|
||
return user_dict(row)
|
||
|
||
|
||
# ---------- Routen: Benutzerverwaltung (nur Admins) ----------
|
||
|
||
|
||
@router.get("/api/users")
|
||
def list_users(_: dict = Depends(require_admin)):
|
||
with transaction() as conn:
|
||
rows = conn.execute("SELECT * FROM users ORDER BY display_name COLLATE NOCASE").fetchall()
|
||
return [user_dict(r) for r in rows]
|
||
|
||
|
||
@router.post("/api/users", status_code=201)
|
||
def create_user(data: dict = Body(...), _: dict = Depends(require_admin)):
|
||
username = _username(data.get("username"))
|
||
display_name = _display_name(data.get("displayName"), username)
|
||
password = _password(data.get("password"))
|
||
with transaction() as conn:
|
||
user_id = _create_user(conn, username, display_name, password, bool(data.get("isAdmin")))
|
||
row = conn.execute("SELECT * FROM users WHERE id = ?", (user_id,)).fetchone()
|
||
return user_dict(row)
|
||
|
||
|
||
def _ensure_other_admin(conn, user_id: int) -> None:
|
||
others = conn.execute("SELECT COUNT(*) FROM users WHERE is_admin = 1 AND id != ?", (user_id,)).fetchone()[0]
|
||
if not others:
|
||
raise HTTPException(400, "Es muss mindestens ein Administrator übrig bleiben.")
|
||
|
||
|
||
@router.patch("/api/users/{user_id}")
|
||
def update_user(user_id: int, data: dict = Body(...), _: dict = Depends(require_admin)):
|
||
with transaction() as conn:
|
||
if not conn.execute("SELECT 1 FROM users WHERE id = ?", (user_id,)).fetchone():
|
||
raise HTTPException(404, "Nicht gefunden.")
|
||
if "username" in data:
|
||
username = _username(data["username"])
|
||
if conn.execute("SELECT 1 FROM users WHERE username = ? AND id != ?", (username, user_id)).fetchone():
|
||
raise HTTPException(400, "Diesen Benutzernamen gibt es schon.")
|
||
conn.execute("UPDATE users SET username = ? WHERE id = ?", (username, user_id))
|
||
if "displayName" in data:
|
||
conn.execute("UPDATE users SET display_name = ? WHERE id = ?", (_display_name(data["displayName"]), user_id))
|
||
if "isAdmin" in data:
|
||
if not data["isAdmin"]:
|
||
_ensure_other_admin(conn, user_id)
|
||
conn.execute("UPDATE users SET is_admin = ? WHERE id = ?", (int(bool(data["isAdmin"])), user_id))
|
||
if data.get("password"):
|
||
conn.execute("UPDATE users SET password_hash = ? WHERE id = ?", (hash_password(_password(data["password"])), user_id))
|
||
conn.execute("DELETE FROM sessions WHERE user_id = ?", (user_id,))
|
||
row = conn.execute("SELECT * FROM users WHERE id = ?", (user_id,)).fetchone()
|
||
return user_dict(row)
|
||
|
||
|
||
@router.delete("/api/users/{user_id}")
|
||
def delete_user(user_id: int, admin: dict = Depends(require_admin)):
|
||
if user_id == admin["id"]:
|
||
raise HTTPException(400, "Das eigene Konto kann nicht gelöscht werden.")
|
||
with transaction() as conn:
|
||
row = conn.execute("SELECT is_admin FROM users WHERE id = ?", (user_id,)).fetchone()
|
||
if not row:
|
||
raise HTTPException(404, "Nicht gefunden.")
|
||
if row["is_admin"]:
|
||
_ensure_other_admin(conn, user_id)
|
||
conn.execute("DELETE FROM users WHERE id = ?", (user_id,))
|
||
return {"ok": True}
|