Wichtel-Werkstatt: Secret-Santa-App mit Räumen, Ausschlüssen und Recovery-Links
- FastAPI + SQLite, serverseitig gerenderte Templates, mobil-zuerst - Admin-Passwort, einseitige Ausschlüsse, eingefrorene Auslosung - Teilnehmer sehen Ergebnis per Cookie; Einmal-Recovery-Links bei Verlust - Docker/podman-tauglich (Entrypoint mit Privilegien-Drop, SELinux-:z) - Unit-Tests für Auslosung, E2E-Testskript (30 Checks)
This commit is contained in:
+496
@@ -0,0 +1,496 @@
|
||||
"""Wichtel-Werkstatt – FastAPI-Anwendung.
|
||||
|
||||
Routen:
|
||||
/ Raum erstellen oder per Code beitreten
|
||||
/r/<code> Teilnehmeransicht (Registrierung / Status / Ergebnis)
|
||||
/r/<code>/admin Admin-Login bzw. Admin-Panel
|
||||
/r/<code>/recover/<token> Einmal-Link, um ein verlorenes Cookie zu ersetzen
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import sqlite3
|
||||
from contextlib import asynccontextmanager
|
||||
from pathlib import Path
|
||||
|
||||
from fastapi import FastAPI, Form, Request
|
||||
from fastapi.responses import HTMLResponse, RedirectResponse
|
||||
from fastapi.staticfiles import StaticFiles
|
||||
from fastapi.templating import Jinja2Templates
|
||||
|
||||
from . import db
|
||||
from .draw import draw_assignment
|
||||
|
||||
BASE_DIR = Path(__file__).resolve().parent
|
||||
COOKIE_MAX_AGE = 2 * 365 * 24 * 60 * 60 # 2 Jahre
|
||||
|
||||
MIN_PASSWORD_LENGTH = 4
|
||||
MIN_NAME_LENGTH = 1
|
||||
MAX_NAME_LENGTH = 50
|
||||
MAX_ROOM_NAME_LENGTH = 60
|
||||
MIN_PARTICIPANTS_FOR_DRAW = 3
|
||||
|
||||
|
||||
@asynccontextmanager
|
||||
async def lifespan(_: FastAPI):
|
||||
db.init_db()
|
||||
yield
|
||||
|
||||
|
||||
app = FastAPI(title="Wichtel-Werkstatt", lifespan=lifespan)
|
||||
app.mount("/static", StaticFiles(directory=BASE_DIR / "static"), name="static")
|
||||
templates = Jinja2Templates(directory=BASE_DIR / "templates")
|
||||
|
||||
|
||||
# ---------- Helpers ----------
|
||||
|
||||
def conn() -> sqlite3.Connection:
|
||||
return db.connect()
|
||||
|
||||
|
||||
def is_secure(request: Request) -> bool:
|
||||
"""Secure-Cookie nur über HTTPS (direkt oder via X-Forwarded-Proto)."""
|
||||
forwarded = request.headers.get("x-forwarded-proto", "")
|
||||
return "https" in forwarded or request.url.scheme == "https"
|
||||
|
||||
|
||||
def set_cookie(response, request: Request, name: str, value: str, path: str) -> None:
|
||||
response.set_cookie(
|
||||
name,
|
||||
value,
|
||||
max_age=COOKIE_MAX_AGE,
|
||||
httponly=True,
|
||||
secure=is_secure(request),
|
||||
samesite="lax",
|
||||
path=path,
|
||||
)
|
||||
|
||||
|
||||
def participant_cookie(room_id: str) -> str:
|
||||
return f"wichtel_p_{room_id}"
|
||||
|
||||
|
||||
def admin_cookie(room_id: str) -> str:
|
||||
return f"wichtel_a_{room_id}"
|
||||
|
||||
|
||||
def base_url(request: Request) -> str:
|
||||
return str(request.base_url).rstrip("/")
|
||||
|
||||
|
||||
def share_url(request: Request, code: str) -> str:
|
||||
return f"{base_url(request)}/r/{code}"
|
||||
|
||||
|
||||
def render_message(request: Request, title: str, text: str, status: int = 200) -> HTMLResponse:
|
||||
return templates.TemplateResponse(
|
||||
request, "message.html", {"title": title, "text": text}, status_code=status
|
||||
)
|
||||
|
||||
|
||||
def room_not_found(request: Request) -> HTMLResponse:
|
||||
return render_message(
|
||||
request,
|
||||
"Raum nicht gefunden 🤔",
|
||||
"Diesen Wichtel-Raum gibt es nicht (mehr). Prüfe den Link oder Code – "
|
||||
"oder frage den Admin nach dem richtigen Einladungslink.",
|
||||
status=404,
|
||||
)
|
||||
|
||||
|
||||
def get_admin_room(request: Request, room) -> bool:
|
||||
"""True, wenn das Admin-Cookie gültig ist."""
|
||||
token = request.cookies.get(admin_cookie(room["id"]))
|
||||
if not token or not room["admin_token_hash"]:
|
||||
return False
|
||||
return db.hash_token(token) == room["admin_token_hash"]
|
||||
|
||||
|
||||
def admin_panel_response(request: Request, room, recovery_link: str | None = None,
|
||||
recovery_for: str | None = None, message: str | None = None,
|
||||
message_kind: str = "ok"):
|
||||
with conn() as c:
|
||||
participants = db.list_participants(c, room["id"])
|
||||
exclusions = db.list_exclusions(c, room["id"])
|
||||
admin_participant = None
|
||||
ptoken = request.cookies.get(participant_cookie(room["id"]))
|
||||
if ptoken:
|
||||
admin_participant = db.get_participant_by_token(c, room["id"], ptoken)
|
||||
drawn = db.room_is_drawn(room)
|
||||
can_draw = (not drawn) and len(participants) >= MIN_PARTICIPANTS_FOR_DRAW
|
||||
return templates.TemplateResponse(
|
||||
request,
|
||||
"admin.html",
|
||||
{
|
||||
"room": room,
|
||||
"participants": participants,
|
||||
"exclusions": exclusions,
|
||||
"share_link": share_url(request, room["code"]),
|
||||
"drawn": drawn,
|
||||
"can_draw": can_draw,
|
||||
"min_draw": MIN_PARTICIPANTS_FOR_DRAW,
|
||||
"recovery_link": recovery_link,
|
||||
"recovery_for": recovery_for,
|
||||
"admin_participant": admin_participant,
|
||||
"message": message,
|
||||
"message_kind": message_kind,
|
||||
"base": base_url(request),
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
# ---------- Startseite ----------
|
||||
|
||||
HOME_ERRORS = {
|
||||
"room_name": "Bitte gib einen Raumnamen an (höchstens 60 Zeichen).",
|
||||
"password": "Das Admin-Passwort muss mindestens 4 Zeichen lang sein.",
|
||||
"code": "Zu diesem Code wurde kein Raum gefunden.",
|
||||
}
|
||||
|
||||
|
||||
@app.get("/", response_class=HTMLResponse)
|
||||
def home(request: Request, error: str | None = None):
|
||||
return templates.TemplateResponse(
|
||||
request, "home.html", {"error": HOME_ERRORS.get(error or "")}
|
||||
)
|
||||
|
||||
|
||||
@app.post("/rooms")
|
||||
def create_room(request: Request, room_name: str = Form(...), password: str = Form(...)):
|
||||
room_name = room_name.strip()
|
||||
if not room_name or len(room_name) > MAX_ROOM_NAME_LENGTH:
|
||||
return RedirectResponse("/?error=room_name", status_code=303)
|
||||
if len(password) < MIN_PASSWORD_LENGTH:
|
||||
return RedirectResponse("/?error=password", status_code=303)
|
||||
with conn() as c:
|
||||
room = db.create_room(c, room_name, password)
|
||||
token = db.new_token()
|
||||
db.set_admin_token(c, room["id"], token)
|
||||
response = RedirectResponse(f"/r/{room['code']}/admin", status_code=303)
|
||||
set_cookie(response, request, admin_cookie(room["id"]), token, f"/r/{room['code']}")
|
||||
return response
|
||||
|
||||
|
||||
@app.post("/join")
|
||||
def join(request: Request, code: str = Form(...)):
|
||||
with conn() as c:
|
||||
room = db.get_room_by_code(c, code)
|
||||
if room is None:
|
||||
return RedirectResponse("/?error=code", status_code=303)
|
||||
return RedirectResponse(f"/r/{room['code']}", status_code=303)
|
||||
|
||||
|
||||
# ---------- Teilnehmeransicht ----------
|
||||
|
||||
ROOM_ERRORS = {
|
||||
"name": "Bitte gib einen Namen an (höchstens 50 Zeichen).",
|
||||
"duplicate": "Dieser Name ist im Raum schon vergeben – wähle bitte einen anderen "
|
||||
"(z. B. mit Nachnamen).",
|
||||
}
|
||||
|
||||
|
||||
@app.get("/r/{code}", response_class=HTMLResponse)
|
||||
def room_page(request: Request, code: str, error: str | None = None):
|
||||
with conn() as c:
|
||||
room = db.get_room_by_code(c, code)
|
||||
if room is None:
|
||||
return room_not_found(request)
|
||||
|
||||
participant = None
|
||||
token = request.cookies.get(participant_cookie(room["id"]))
|
||||
if token:
|
||||
participant = db.get_participant_by_token(c, room["id"], token)
|
||||
|
||||
assignment = None
|
||||
if participant and db.room_is_drawn(room):
|
||||
assignment = db.get_assignment_for(c, room["id"], participant["id"])
|
||||
|
||||
return templates.TemplateResponse(
|
||||
request,
|
||||
"room.html",
|
||||
{
|
||||
"room": room,
|
||||
"participant": participant,
|
||||
"assignment": assignment,
|
||||
"drawn": db.room_is_drawn(room),
|
||||
"error": ROOM_ERRORS.get(error or ""),
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
@app.post("/r/{code}/register")
|
||||
def register(request: Request, code: str, name: str = Form(...)):
|
||||
with conn() as c:
|
||||
room = db.get_room_by_code(c, code)
|
||||
if room is None:
|
||||
return room_not_found(request)
|
||||
if db.room_is_drawn(room):
|
||||
return render_message(
|
||||
request,
|
||||
"Zu spät 🎄",
|
||||
"In diesem Raum wurde bereits ausgelost – eine Teilnahme ist nicht mehr möglich.",
|
||||
)
|
||||
|
||||
name = name.strip()
|
||||
if len(name) < MIN_NAME_LENGTH or len(name) > MAX_NAME_LENGTH:
|
||||
return RedirectResponse(f"/r/{room['code']}?error=name", status_code=303)
|
||||
|
||||
token = db.new_token()
|
||||
try:
|
||||
participant = db.add_participant(c, room["id"], name, token)
|
||||
except sqlite3.IntegrityError:
|
||||
return RedirectResponse(f"/r/{room['code']}?error=duplicate", status_code=303)
|
||||
|
||||
response = RedirectResponse(f"/r/{room['code']}", status_code=303)
|
||||
set_cookie(response, request, participant_cookie(room["id"]), token, f"/r/{room['code']}")
|
||||
return response
|
||||
|
||||
|
||||
# ---------- Admin ----------
|
||||
|
||||
@app.get("/r/{code}/admin", response_class=HTMLResponse)
|
||||
def admin_page(request: Request, code: str):
|
||||
with conn() as c:
|
||||
room = db.get_room_by_code(c, code)
|
||||
if room is None:
|
||||
return room_not_found(request)
|
||||
if not get_admin_room(request, room):
|
||||
return templates.TemplateResponse(
|
||||
request, "admin_login.html", {"room": room, "error": None}
|
||||
)
|
||||
return admin_panel_response(request, room)
|
||||
|
||||
|
||||
@app.post("/r/{code}/admin/login")
|
||||
def admin_login(request: Request, code: str, password: str = Form(...)):
|
||||
with conn() as c:
|
||||
room = db.get_room_by_code(c, code)
|
||||
if room is None:
|
||||
return room_not_found(request)
|
||||
if not db.verify_password(password, room["password_hash"]):
|
||||
return templates.TemplateResponse(
|
||||
request, "admin_login.html", {"room": room, "error": True}
|
||||
)
|
||||
token = db.new_token()
|
||||
db.set_admin_token(c, room["id"], token)
|
||||
response = RedirectResponse(f"/r/{room['code']}/admin", status_code=303)
|
||||
set_cookie(response, request, admin_cookie(room["id"]), token, f"/r/{room['code']}")
|
||||
return response
|
||||
|
||||
|
||||
@app.post("/r/{code}/admin/logout")
|
||||
def admin_logout(request: Request, code: str):
|
||||
with conn() as c:
|
||||
room = db.get_room_by_code(c, code)
|
||||
if room is None:
|
||||
return room_not_found(request)
|
||||
db.clear_admin_token(c, room["id"])
|
||||
response = RedirectResponse(f"/r/{room['code']}/admin", status_code=303)
|
||||
response.delete_cookie(admin_cookie(room["id"]), path=f"/r/{room['code']}")
|
||||
return response
|
||||
|
||||
|
||||
@app.post("/r/{code}/admin/self-register")
|
||||
def admin_self_register(request: Request, code: str, name: str = Form(...)):
|
||||
"""Der Admin nimmt selbst am Wichteln teil: legt ihn als Teilnehmer an
|
||||
und setzt zusätzlich den Teilnehmer-Cookie in seinem Browser."""
|
||||
with conn() as c:
|
||||
room = db.get_room_by_code(c, code)
|
||||
if room is None:
|
||||
return room_not_found(request)
|
||||
if not get_admin_room(request, room):
|
||||
return RedirectResponse(f"/r/{room['code']}/admin", status_code=303)
|
||||
if db.room_is_drawn(room):
|
||||
return admin_panel_response(
|
||||
request, room,
|
||||
message="Nach der Auslosung ist keine Teilnahme mehr möglich.",
|
||||
message_kind="error",
|
||||
)
|
||||
name = name.strip()
|
||||
if len(name) < MIN_NAME_LENGTH or len(name) > MAX_NAME_LENGTH:
|
||||
return admin_panel_response(
|
||||
request, room,
|
||||
message="Bitte gib einen gültigen Namen an (höchstens 50 Zeichen).",
|
||||
message_kind="error",
|
||||
)
|
||||
token = db.new_token()
|
||||
try:
|
||||
db.add_participant(c, room["id"], name, token)
|
||||
except sqlite3.IntegrityError:
|
||||
return admin_panel_response(
|
||||
request, room, message="Dieser Name ist bereits vergeben.",
|
||||
message_kind="error",
|
||||
)
|
||||
response = admin_panel_response(
|
||||
request, room,
|
||||
message=f"Schön, dass du dabei bist: Du wichtelst als »{name}« mit! 🎅 "
|
||||
"Deine Teilnehmeransicht (und später dein Ergebnis) erreichst du "
|
||||
"über den Einladungslink.",
|
||||
message_kind="ok",
|
||||
)
|
||||
set_cookie(response, request, participant_cookie(room["id"]), token, f"/r/{room['code']}")
|
||||
return response
|
||||
|
||||
|
||||
@app.post("/r/{code}/participants/{pid}/delete")
|
||||
def delete_participant(request: Request, code: str, pid: str):
|
||||
with conn() as c:
|
||||
room = db.get_room_by_code(c, code)
|
||||
if room is None:
|
||||
return room_not_found(request)
|
||||
if not get_admin_room(request, room):
|
||||
return RedirectResponse(f"/r/{room['code']}/admin", status_code=303)
|
||||
if db.room_is_drawn(room):
|
||||
return admin_panel_response(
|
||||
request, room, message="Nach der Auslosung können keine Teilnehmer mehr gelöscht werden.",
|
||||
message_kind="error",
|
||||
)
|
||||
db.delete_participant(c, room["id"], pid)
|
||||
return RedirectResponse(f"/r/{room['code']}/admin", status_code=303)
|
||||
|
||||
|
||||
@app.post("/r/{code}/exclusions/add")
|
||||
def add_exclusion(request: Request, code: str, from_id: str = Form(...), to_id: str = Form(...)):
|
||||
with conn() as c:
|
||||
room = db.get_room_by_code(c, code)
|
||||
if room is None:
|
||||
return room_not_found(request)
|
||||
if not get_admin_room(request, room):
|
||||
return RedirectResponse(f"/r/{room['code']}/admin", status_code=303)
|
||||
if db.room_is_drawn(room):
|
||||
return admin_panel_response(
|
||||
request, room, message="Nach der Auslosung können keine Ausschlüsse mehr geändert werden.",
|
||||
message_kind="error",
|
||||
)
|
||||
if from_id == to_id:
|
||||
return admin_panel_response(
|
||||
request, room, message="Ein Teilnehmer kann sich nicht selbst ausgeschlossen werden.",
|
||||
message_kind="error",
|
||||
)
|
||||
db.add_exclusion(c, room["id"], from_id, to_id)
|
||||
return RedirectResponse(f"/r/{room['code']}/admin", status_code=303)
|
||||
|
||||
|
||||
@app.post("/r/{code}/exclusions/{from_id}/{to_id}/delete")
|
||||
def delete_exclusion(request: Request, code: str, from_id: str, to_id: str):
|
||||
with conn() as c:
|
||||
room = db.get_room_by_code(c, code)
|
||||
if room is None:
|
||||
return room_not_found(request)
|
||||
if not get_admin_room(request, room):
|
||||
return RedirectResponse(f"/r/{room['code']}/admin", status_code=303)
|
||||
if db.room_is_drawn(room):
|
||||
return admin_panel_response(
|
||||
request, room, message="Nach der Auslosung können keine Ausschlüsse mehr geändert werden.",
|
||||
message_kind="error",
|
||||
)
|
||||
db.remove_exclusion(c, room["id"], from_id, to_id)
|
||||
return RedirectResponse(f"/r/{room['code']}/admin", status_code=303)
|
||||
|
||||
|
||||
@app.post("/r/{code}/draw")
|
||||
def draw(request: Request, code: str):
|
||||
with conn() as c:
|
||||
room = db.get_room_by_code(c, code)
|
||||
if room is None:
|
||||
return room_not_found(request)
|
||||
if not get_admin_room(request, room):
|
||||
return RedirectResponse(f"/r/{room['code']}/admin", status_code=303)
|
||||
if db.room_is_drawn(room):
|
||||
return admin_panel_response(
|
||||
request, room, message="Es wurde bereits ausgelost.", message_kind="error"
|
||||
)
|
||||
participants = db.list_participants(c, room["id"])
|
||||
if len(participants) < MIN_PARTICIPANTS_FOR_DRAW:
|
||||
return admin_panel_response(
|
||||
request, room,
|
||||
message=f"Mindestens {MIN_PARTICIPANTS_FOR_DRAW} Teilnehmer nötig – aktuell sind es {len(participants)}.",
|
||||
message_kind="error",
|
||||
)
|
||||
ids = [p["id"] for p in participants]
|
||||
mapping = draw_assignment(ids, db.exclusion_pairs(c, room["id"]))
|
||||
if mapping is None:
|
||||
return admin_panel_response(
|
||||
request, room,
|
||||
message="Mit diesen Ausschlüssen ist keine gültige Auslosung möglich. "
|
||||
"Bitte entferne oder lockere Ausschlüsse.",
|
||||
message_kind="error",
|
||||
)
|
||||
db.save_assignments(c, room["id"], mapping)
|
||||
room = db.get_room_by_id(c, room["id"])
|
||||
return admin_panel_response(
|
||||
request, room,
|
||||
message="🎉 Die Auslosung ist abgeschlossen! Die Teilnehmer sehen ihr Ergebnis, "
|
||||
"wenn sie ihren Einladungslink öffnen.",
|
||||
message_kind="ok",
|
||||
)
|
||||
|
||||
|
||||
# ---------- Recovery-Links (Cookie verloren) ----------
|
||||
|
||||
@app.post("/r/{code}/recovery/{pid}")
|
||||
def create_recovery_link(request: Request, code: str, pid: str):
|
||||
"""Erzeugt einen Einmal-Link, mit dem der Teilnehmer sein Cookie zurückbekommt.
|
||||
|
||||
Bewusst auch nach der Auslosung erlaubt: Der Admin sieht dabei weiterhin
|
||||
nicht, wer wen gezogen hat – der Link stellt nur die Sicht des Teilnehmers
|
||||
selbst wieder her.
|
||||
"""
|
||||
with conn() as c:
|
||||
room = db.get_room_by_code(c, code)
|
||||
if room is None:
|
||||
return room_not_found(request)
|
||||
if not get_admin_room(request, room):
|
||||
return RedirectResponse(f"/r/{room['code']}/admin", status_code=303)
|
||||
participant = db.get_participant(c, pid)
|
||||
if participant is None or participant["room_id"] != room["id"]:
|
||||
return admin_panel_response(
|
||||
request, room, message="Teilnehmer nicht gefunden.", message_kind="error"
|
||||
)
|
||||
token = db.new_token()
|
||||
db.create_recovery_link(c, room["id"], pid, token)
|
||||
link = f"{share_url(request, room['code'])}/recover/{token}"
|
||||
return admin_panel_response(
|
||||
request, room, recovery_link=link, recovery_for=participant["name"],
|
||||
message=None,
|
||||
)
|
||||
|
||||
|
||||
@app.get("/r/{code}/recover/{token}", response_class=HTMLResponse)
|
||||
def redeem_recovery_link(request: Request, code: str, token: str):
|
||||
with conn() as c:
|
||||
room = db.get_room_by_code(c, code)
|
||||
if room is None:
|
||||
return room_not_found(request)
|
||||
link = db.get_recovery_link(c, token)
|
||||
if link is None or link["room_id"] != room["id"]:
|
||||
return render_message(
|
||||
request,
|
||||
"Link ungültig 🔒",
|
||||
"Dieser Wiederherstellungslink ist ungültig. Bitte den Admin um einen neuen Link.",
|
||||
status=404,
|
||||
)
|
||||
if link["used_at"] is not None:
|
||||
return render_message(
|
||||
request,
|
||||
"Link bereits verwendet 🔒",
|
||||
"Dieser Wiederherstellungslink wurde schon benutzt. "
|
||||
"Falls du erneut Zugriff brauchst, bitte den Admin um einen neuen Link.",
|
||||
)
|
||||
new_token = db.new_token()
|
||||
db.set_participant_token(c, link["participant_id"], new_token)
|
||||
db.mark_recovery_link_used(c, link["token_hash"])
|
||||
|
||||
response = RedirectResponse(f"/r/{room['code']}", status_code=303)
|
||||
set_cookie(response, request, participant_cookie(room["id"]), new_token, f"/r/{room['code']}")
|
||||
return response
|
||||
|
||||
|
||||
@app.exception_handler(sqlite3.Error)
|
||||
def sqlite_error(request: Request, exc: sqlite3.Error):
|
||||
return render_message(
|
||||
request,
|
||||
"Ups ❄",
|
||||
"Es ist ein Fehler beim Speichern aufgetreten. Bitte versuche es noch einmal.",
|
||||
status=500,
|
||||
)
|
||||
Reference in New Issue
Block a user