220 lines
8.2 KiB
Python
220 lines
8.2 KiB
Python
"""Validierung und Plausibilitätsprüfung von Highscore-Einreichungen.
|
||
|
||
Reine Funktionen ohne Flask-Abhängigkeit, damit sie isoliert testbar sind.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
import re
|
||
import unicodedata
|
||
from collections.abc import Iterable, Mapping
|
||
from dataclasses import dataclass
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
NAME_MAX_LENGTH = 16
|
||
|
||
#: Obergrenze für die Rohlänge, bevor überhaupt bereinigt wird – schützt vor
|
||
#: unnötiger Arbeit an absurd langen Eingaben.
|
||
NAME_RAW_MAX_LENGTH = 200
|
||
|
||
#: Alles außer Buchstaben, Ziffern, Unterstrich, Leerzeichen und ein paar harmlosen
|
||
#: Satzzeichen wird entfernt. Umlaute und andere Unicode-Buchstaben bleiben erhalten.
|
||
_DISALLOWED_NAME_CHARS = re.compile(r"[^\w .\-'!?]")
|
||
|
||
RUN_TOKEN_MAX_LENGTH = 512
|
||
|
||
|
||
class SubmissionError(ValueError):
|
||
"""Ungültige Einreichung; ``fields`` ordnet Feldnamen eine lesbare Meldung zu."""
|
||
|
||
def __init__(self, fields: dict[str, str]) -> None:
|
||
super().__init__("; ".join(f"{key}: {msg}" for key, msg in fields.items()))
|
||
self.fields = fields
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class ScoreSubmission:
|
||
"""Eine formal gültige Highscore-Einreichung."""
|
||
|
||
name: str
|
||
score: int
|
||
wave: int
|
||
duration_s: int
|
||
run_token: str
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class PlausibilityRules:
|
||
"""Grenzwerte für die Plausibilitätsprüfung (siehe ``config.GAME_RULE_DEFAULTS``)."""
|
||
|
||
score_max: int
|
||
score_flat_allowance: int
|
||
score_max_points_per_second: int
|
||
wave_max: int
|
||
wave_min_seconds: int
|
||
duration_max_seconds: int
|
||
duration_tolerance_seconds: int
|
||
|
||
@classmethod
|
||
def from_config(cls, config: Mapping[str, Any]) -> PlausibilityRules:
|
||
"""Erzeugt die Regeln aus einer Flask-Konfiguration."""
|
||
return cls(
|
||
score_max=int(config["SCORE_MAX"]),
|
||
score_flat_allowance=int(config["SCORE_FLAT_ALLOWANCE"]),
|
||
score_max_points_per_second=int(config["SCORE_MAX_POINTS_PER_SECOND"]),
|
||
wave_max=int(config["WAVE_MAX"]),
|
||
wave_min_seconds=int(config["WAVE_MIN_SECONDS"]),
|
||
duration_max_seconds=int(config["DURATION_MAX_SECONDS"]),
|
||
duration_tolerance_seconds=int(config["DURATION_TOLERANCE_SECONDS"]),
|
||
)
|
||
|
||
|
||
def sanitize_name(raw: str) -> str:
|
||
"""Bereinigt einen Spielernamen für Anzeige und Speicherung.
|
||
|
||
Schritte: Unicode-Normalisierung (NFKC, macht z. B. Vollbreiten-Zeichen zu normalen),
|
||
Entfernen von Steuer- und Formatzeichen (u. a. unsichtbare Zero-Width- und
|
||
Bidi-Zeichen, mit denen man Namen optisch fälschen könnte), Entfernen nicht
|
||
erlaubter Zeichen und Zusammenfassen von Leerraum.
|
||
|
||
Die Ausgabe im Frontend wird trotzdem immer als Text (nicht als HTML) gesetzt –
|
||
die Bereinigung ist eine zweite Verteidigungslinie, nicht die einzige.
|
||
"""
|
||
text = unicodedata.normalize("NFKC", raw)
|
||
# Kategorie „C*“ = Steuer-, Format-, Surrogat-, private und nicht zugewiesene Zeichen.
|
||
text = "".join(ch for ch in text if not unicodedata.category(ch).startswith("C"))
|
||
text = _DISALLOWED_NAME_CHARS.sub("", text)
|
||
return " ".join(text.split())
|
||
|
||
|
||
def normalize_for_blocklist(text: str) -> str:
|
||
"""Vergleichsform für die Sperrliste: klein geschrieben, nur Buchstaben und Ziffern.
|
||
|
||
So lassen sich gesperrte Begriffe nicht mit Leerzeichen, Punkten oder anderer
|
||
Groß-/Kleinschreibung umgehen („M.a.x Muster“ entspricht „maxmuster“).
|
||
"""
|
||
folded = unicodedata.normalize("NFKC", text).casefold()
|
||
return "".join(ch for ch in folded if ch.isalnum())
|
||
|
||
|
||
def load_blocklist(path: str | Path) -> list[str]:
|
||
"""Liest die Sperrliste (UTF-8, ein Begriff pro Zeile, ``#`` leitet Kommentare ein).
|
||
|
||
Fehlt die Datei, ist nichts gesperrt. Ist sie unlesbar, wird gewarnt, statt die
|
||
Einreichung scheitern zu lassen.
|
||
"""
|
||
try:
|
||
text = Path(path).read_text(encoding="utf-8")
|
||
except FileNotFoundError:
|
||
return []
|
||
except (OSError, UnicodeDecodeError) as exc:
|
||
logger.warning("Sperrliste %s konnte nicht gelesen werden: %s", path, exc)
|
||
return []
|
||
terms = []
|
||
for line in text.splitlines():
|
||
entry = line.strip()
|
||
if entry and not entry.startswith("#"):
|
||
normalized = normalize_for_blocklist(entry)
|
||
if normalized:
|
||
terms.append(normalized)
|
||
return terms
|
||
|
||
|
||
def is_name_blocked(name: str, terms: Iterable[str]) -> bool:
|
||
"""True, wenn ein gesperrter Begriff im (normalisierten) Namen vorkommt."""
|
||
normalized = normalize_for_blocklist(name)
|
||
return any(term in normalized for term in terms)
|
||
|
||
|
||
def _strict_int(value: object) -> int | None:
|
||
"""Nur echte Ganzzahlen: ``True`` ist in Python auch ein ``int``, aber kein Punktestand."""
|
||
if isinstance(value, bool) or not isinstance(value, int):
|
||
return None
|
||
return value
|
||
|
||
|
||
def parse_submission(payload: Mapping[str, Any], rules: PlausibilityRules) -> ScoreSubmission:
|
||
"""Prüft Typen und Wertebereiche einer Einreichung.
|
||
|
||
Raises:
|
||
SubmissionError: mit allen gefundenen Fehlern auf einmal, damit das Frontend
|
||
nicht Fehler für Fehler nachfragen muss.
|
||
"""
|
||
errors: dict[str, str] = {}
|
||
|
||
raw_name = payload.get("name")
|
||
name = ""
|
||
if not isinstance(raw_name, str):
|
||
errors["name"] = "Name fehlt oder ist kein Text."
|
||
elif len(raw_name) > NAME_RAW_MAX_LENGTH:
|
||
errors["name"] = f"Name darf höchstens {NAME_MAX_LENGTH} Zeichen lang sein."
|
||
else:
|
||
name = sanitize_name(raw_name)
|
||
if not name:
|
||
errors["name"] = "Name ist leer oder enthält nur unzulässige Zeichen."
|
||
elif len(name) > NAME_MAX_LENGTH:
|
||
errors["name"] = f"Name darf höchstens {NAME_MAX_LENGTH} Zeichen lang sein."
|
||
|
||
score = _strict_int(payload.get("score"))
|
||
if score is None or not 0 <= score <= rules.score_max:
|
||
errors["score"] = f"Punkte müssen eine ganze Zahl zwischen 0 und {rules.score_max} sein."
|
||
|
||
wave = _strict_int(payload.get("wave"))
|
||
if wave is None or not 1 <= wave <= rules.wave_max:
|
||
errors["wave"] = f"Welle muss eine ganze Zahl zwischen 1 und {rules.wave_max} sein."
|
||
|
||
duration_s = _strict_int(payload.get("duration_s"))
|
||
if duration_s is None or not 0 <= duration_s <= rules.duration_max_seconds:
|
||
errors["duration_s"] = (
|
||
f"Spieldauer muss eine ganze Zahl zwischen 0 und {rules.duration_max_seconds} sein."
|
||
)
|
||
|
||
run_token = payload.get("run_token")
|
||
if not isinstance(run_token, str) or not 0 < len(run_token) <= RUN_TOKEN_MAX_LENGTH:
|
||
errors["run_token"] = "Run-Token fehlt oder ist ungültig." # noqa: S105
|
||
|
||
if errors:
|
||
raise SubmissionError(errors)
|
||
|
||
# Nach der Prüfung oben sind alle Werte gesetzt; die asserts helfen nur dem Typprüfer.
|
||
assert score is not None and wave is not None and duration_s is not None # noqa: S101
|
||
assert isinstance(run_token, str) # noqa: S101
|
||
return ScoreSubmission(
|
||
name=name, score=score, wave=wave, duration_s=duration_s, run_token=run_token
|
||
)
|
||
|
||
|
||
def check_plausibility(
|
||
submission: ScoreSubmission, elapsed_s: float, rules: PlausibilityRules
|
||
) -> str | None:
|
||
"""Prüft, ob Punktestand, Welle und Dauer zusammenpassen können.
|
||
|
||
Args:
|
||
submission: Formal gültige Einreichung.
|
||
elapsed_s: Serverseitig gemessene Zeit seit Ausgabe des Run-Tokens.
|
||
rules: Grenzwerte.
|
||
|
||
Returns:
|
||
``None`` wenn plausibel, sonst eine kurze deutsche Begründung.
|
||
"""
|
||
# Die gemeldete (aktive) Spielzeit kann kürzer sein als die Token-Zeit, weil Pausen
|
||
# nicht zählen – aber nie länger.
|
||
if submission.duration_s > elapsed_s + rules.duration_tolerance_seconds:
|
||
return "Die gemeldete Spieldauer ist länger als die tatsächliche Rundenzeit."
|
||
|
||
# Jede überstandene Welle kostet mindestens die Pause bis zur nächsten.
|
||
min_duration = (submission.wave - 1) * rules.wave_min_seconds
|
||
if submission.duration_s + rules.duration_tolerance_seconds < min_duration:
|
||
return "Diese Welle ist in der gemeldeten Zeit nicht erreichbar."
|
||
|
||
max_score = (
|
||
rules.score_flat_allowance + submission.duration_s * rules.score_max_points_per_second
|
||
)
|
||
if submission.score > max_score:
|
||
return "Der Punktestand ist für die Spieldauer nicht plausibel."
|
||
return None
|