"""SQLite-Zugriff: Schema, Highscores, Rate-Limit-Protokoll und verbrauchte Run-Tokens. SQLite reicht hier völlig aus und braucht auf Shared Hosting keinen separaten Datenbankserver. Weil Passenger mehrere Prozesse parallel betreibt, liegt auch der Zustand für Rate-Limiting und Einmal-Tokens in der Datenbank statt im Arbeitsspeicher. """ from __future__ import annotations import math import sqlite3 import unicodedata from collections.abc import Iterable, Iterator from contextlib import contextmanager from datetime import datetime, timezone from pathlib import Path from typing import Any import click from flask import Flask, current_app, g from flask.cli import AppGroup, with_appcontext SCHEMA_VERSION = 1 #: Wie lange ein Prozess auf eine gesperrte Datenbank wartet, bevor er aufgibt. BUSY_TIMEOUT_SECONDS = 10.0 SCHEMA = """ CREATE TABLE IF NOT EXISTS highscores ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL, score INTEGER NOT NULL CHECK (score >= 0), wave INTEGER NOT NULL CHECK (wave >= 1), duration_s INTEGER NOT NULL CHECK (duration_s >= 0), created_at TEXT NOT NULL ); CREATE INDEX IF NOT EXISTS idx_highscores_rank ON highscores (score DESC, id ASC); -- Ein Eintrag pro angenommenem Einreichungsversuch; client_hash ist ein HMAC der -- IP-Adresse, damit keine IP-Adressen im Klartext gespeichert werden. CREATE TABLE IF NOT EXISTS submission_log ( id INTEGER PRIMARY KEY AUTOINCREMENT, client_hash TEXT NOT NULL, created_at REAL NOT NULL ); CREATE INDEX IF NOT EXISTS idx_submission_log_client ON submission_log (client_hash, created_at); -- Bereits eingelöste Run-Tokens: jede Spielrunde darf nur einen Highscore erzeugen. CREATE TABLE IF NOT EXISTS used_run_tokens ( run_id TEXT PRIMARY KEY, used_at REAL NOT NULL ); """ def connect(db_path: str | Path) -> sqlite3.Connection: """Öffnet eine Verbindung im Autocommit-Modus; Transaktionen werden explizit gesteuert.""" conn = sqlite3.connect(str(db_path), timeout=BUSY_TIMEOUT_SECONDS, isolation_level=None) conn.row_factory = sqlite3.Row conn.execute("PRAGMA foreign_keys = ON") return conn @contextmanager def immediate_transaction(db: sqlite3.Connection) -> Iterator[sqlite3.Connection]: """Transaktion, die die Schreibsperre sofort holt (``BEGIN IMMEDIATE``). So kann zwischen Lesen und Schreiben kein anderer Prozess dazwischenfunken; bei einer Ausnahme wird alles zurückgerollt. """ db.execute("BEGIN IMMEDIATE") try: yield db except BaseException: db.execute("ROLLBACK") raise db.execute("COMMIT") def init_db(db_path: str | Path) -> None: """Legt Datenbankdatei, Elternordner und Tabellen an. Mehrfaches Ausführen ist harmlos.""" path = Path(db_path) path.parent.mkdir(parents=True, exist_ok=True) conn = connect(path) try: conn.executescript(SCHEMA) conn.execute(f"PRAGMA user_version = {SCHEMA_VERSION:d}") finally: conn.close() def get_db() -> sqlite3.Connection: """Liefert die Verbindung des aktuellen Requests (wird am Request-Ende geschlossen).""" if "db" not in g: g.db = connect(current_app.config["DATABASE_PATH"]) return g.db def close_db(_exc: BaseException | None = None) -> None: """Schließt die Request-Verbindung, falls eine geöffnet wurde.""" db = g.pop("db", None) if db is not None: db.close() @click.command("init-db") @with_appcontext def init_db_command() -> None: """Flask-CLI: ``flask init-db`` legt die Datenbank an (auch nachträglich).""" init_db(current_app.config["DATABASE_PATH"]) click.echo(f"Datenbank bereit: {current_app.config['DATABASE_PATH']}") def init_app(app: Flask) -> None: """Verdrahtet die Datenbank mit der App und stellt sicher, dass das Schema existiert.""" init_db(app.config["DATABASE_PATH"]) app.teardown_appcontext(close_db) app.cli.add_command(init_db_command) app.cli.add_command(scores_cli) # --------------------------------------------------------------------------- Highscores def top_scores(db: sqlite3.Connection, limit: int) -> list[dict[str, Any]]: """Liefert die besten ``limit`` Einträge; bei Gleichstand gewinnt der ältere Eintrag.""" rows = db.execute( "SELECT name, score, wave, created_at FROM highscores ORDER BY score DESC, id ASC LIMIT ?", (limit,), ).fetchall() return [ { "rank": rank, "name": row["name"], "score": row["score"], "wave": row["wave"], "created_at": row["created_at"], } for rank, row in enumerate(rows, start=1) ] def insert_score( db: sqlite3.Connection, name: str, score: int, wave: int, duration_s: int, now: datetime | None = None, ) -> int: """Speichert einen Highscore und gibt dessen ID zurück.""" created_at = (now or datetime.now(timezone.utc)).isoformat(timespec="seconds") cursor = db.execute( "INSERT INTO highscores (name, score, wave, duration_s, created_at) VALUES (?, ?, ?, ?, ?)", (name, score, wave, duration_s, created_at), ) return int(cursor.lastrowid or 0) def rank_of(db: sqlite3.Connection, score_id: int) -> int: """Platzierung eines Eintrags nach derselben Sortierung wie :func:`top_scores`.""" row = db.execute( "SELECT 1 + COUNT(*) AS rank FROM highscores AS other, " "(SELECT score, id FROM highscores WHERE id = ?) AS mine " "WHERE other.score > mine.score OR (other.score = mine.score AND other.id < mine.id)", (score_id,), ).fetchone() return int(row["rank"]) # ------------------------------------------------------------------------- Moderation def list_scores(db: sqlite3.Connection, limit: int) -> list[dict[str, Any]]: """Einträge mit ID für die Moderation (gleiche Sortierung wie die Ehrentafel).""" rows = db.execute( "SELECT id, name, score, wave, duration_s, created_at FROM highscores " "ORDER BY score DESC, id ASC LIMIT ?", (limit,), ).fetchall() return [dict(row) for row in rows] def delete_scores(db: sqlite3.Connection, score_ids: Iterable[int]) -> int: """Löscht Einträge per ID und gibt die Anzahl gelöschter Einträge zurück.""" ids = [int(score_id) for score_id in score_ids] if not ids: return 0 placeholders = ", ".join("?" for _ in ids) cursor = db.execute(f"DELETE FROM highscores WHERE id IN ({placeholders})", ids) # noqa: S608 – nur Platzhalter return cursor.rowcount def delete_scores_by_name(db: sqlite3.Connection, name: str) -> int: """Löscht alle Einträge mit genau diesem Namen (ohne Rücksicht auf Groß-/Kleinschreibung). Verglichen wird in Python: SQLites ``COLLATE NOCASE`` beachtet nur ASCII, „Ü“ und „ü“ wären dort verschieden. """ def fold(text: str) -> str: return unicodedata.normalize("NFKC", text).strip().casefold() target = fold(name) rows = db.execute("SELECT id, name FROM highscores").fetchall() return delete_scores(db, [row["id"] for row in rows if fold(row["name"]) == target]) #: CLI-Gruppe ``flask scores …``. Kommandos einer AppGroup laufen automatisch im App-Kontext. scores_cli = AppGroup("scores", help="Ehrentafel moderieren (auflisten, Einträge löschen).") @scores_cli.command("list") @click.option("--limit", default=50, show_default=True, type=click.IntRange(1, 100_000)) def scores_list_command(limit: int) -> None: """Zeigt Einträge mit ID, Punkten, Welle, Datum und Name.""" rows = list_scores(get_db(), limit) if not rows: click.echo("Keine Einträge.") return click.echo(f"{'ID':>6} {'Punkte':>9} {'Welle':>5} {'Datum (UTC)':<25} Name") for row in rows: numbers = f"{row['id']:>6} {row['score']:>9} {row['wave']:>5}" click.echo(f"{numbers} {row['created_at']:<25} {row['name']}") @scores_cli.command("delete") @click.argument("score_ids", nargs=-1, required=True, type=int) def scores_delete_command(score_ids: tuple[int, ...]) -> None: """Löscht Einträge anhand ihrer ID (siehe ``flask scores list``).""" deleted = delete_scores(get_db(), score_ids) click.echo(f"{deleted} Eintrag/Einträge gelöscht.") @scores_cli.command("delete-name") @click.argument("name") def scores_delete_name_command(name: str) -> None: """Löscht alle Einträge mit diesem Namen.""" deleted = delete_scores_by_name(get_db(), name) click.echo(f"{deleted} Eintrag/Einträge mit dem Namen {name!r} gelöscht.") # ------------------------------------------------------------------------- Rate-Limit def register_submission_attempt( db: sqlite3.Connection, client_hash: str, now: float, window_s: int, max_attempts: int, ) -> tuple[bool, int]: """Prüft das Rate-Limit (gleitendes Zeitfenster) und protokolliert den Versuch. Zählen und Eintragen passieren in einer ``BEGIN IMMEDIATE``-Transaktion, damit zwei parallele Passenger-Prozesse das Limit nicht gemeinsam überschreiten können. Abgelehnte Versuche werden nicht protokolliert, sonst würde ein blockierter Client sein eigenes Zeitfenster endlos verlängern. Returns: ``(erlaubt, retry_after_s)``; ``retry_after_s`` ist 0, wenn erlaubt. """ with immediate_transaction(db): # Aufräumen im selben Zug hält die Tabelle klein, ganz ohne Cronjob. db.execute("DELETE FROM submission_log WHERE created_at < ?", (now - window_s,)) rows = db.execute( "SELECT created_at FROM submission_log WHERE client_hash = ? ORDER BY created_at ASC", (client_hash,), ).fetchall() if len(rows) >= max_attempts: # Ein Platz wird frei, sobald der älteste Versuch aus dem Fenster fällt. oldest = float(rows[0]["created_at"]) return False, max(1, math.ceil(oldest + window_s - now)) db.execute( "INSERT INTO submission_log (client_hash, created_at) VALUES (?, ?)", (client_hash, now), ) return True, 0 # ------------------------------------------------------------------------- Run-Tokens def consume_run_token(db: sqlite3.Connection, run_id: str, now: float, max_age_s: int) -> bool: """Markiert ein Run-Token als eingelöst. Returns: ``False``, wenn das Token schon einmal eingelöst wurde. """ # Abgelaufene Tokens scheitern ohnehin an der Signaturprüfung und können weg. db.execute("DELETE FROM used_run_tokens WHERE used_at < ?", (now - max_age_s,)) cursor = db.execute( "INSERT OR IGNORE INTO used_run_tokens (run_id, used_at) VALUES (?, ?)", (run_id, now), ) return cursor.rowcount == 1