slop-game-opus5.5/app/api.py

"""JSON-API: Highscores lesen/einreichen und Run-Tokens ausgeben.

Ablauf einer Einreichung:
1. Beim Spielstart holt das Frontend per ``POST /api/runs`` ein signiertes Run-Token.
2. Nach dem Game Over schickt es Name, Punkte, Welle, Spieldauer und das Token.
3. Der Server prüft Rate-Limit, Format, Token (Signatur, Alter, Einmaligkeit) und
   Plausibilität, bevor er speichert.
"""

from __future__ import annotations

import hashlib
import hmac
import time
import uuid
from datetime import datetime, timezone

from flask import Blueprint, Response, current_app, jsonify, request
from itsdangerous import BadSignature, SignatureExpired, URLSafeTimedSerializer

from .config import ensure_secret_key
from .models import (
    consume_run_token,
    get_db,
    immediate_transaction,
    insert_score,
    rank_of,
    register_submission_attempt,
    top_scores,
)
from .validation import (
    PlausibilityRules,
    SubmissionError,
    check_plausibility,
    is_name_blocked,
    load_blocklist,
    parse_submission,
)

bp = Blueprint("api", __name__, url_prefix="/api")

#: Salt trennt Run-Tokens von anderen Signaturen, die denselben SECRET_KEY nutzen könnten.
RUN_TOKEN_SALT = "glory-defense.run-token.v1"  # noqa: S105 (Salt, kein Passwort)


def _now() -> float:
    """Aktuelle Unix-Zeit; eigene Funktion, damit Tests die Uhr gezielt verstellen können."""
    return time.time()


def _serializer() -> URLSafeTimedSerializer:
    return URLSafeTimedSerializer(ensure_secret_key(current_app.config), salt=RUN_TOKEN_SALT)


def _client_hash() -> str:
    """HMAC der Client-IP: genug, um Clients zu unterscheiden, ohne IPs zu speichern."""
    address = request.remote_addr or "unknown"
    key = ensure_secret_key(current_app.config).encode("utf-8")
    return hmac.new(key, address.encode("utf-8"), hashlib.sha256).hexdigest()


def error_response(status: int, code: str, message: str, **extra: object) -> tuple[Response, int]:
    """Einheitliches Fehlerformat: ``{"error": code, "message": text, ...}``."""
    return jsonify({"error": code, "message": message, **extra}), status


def _no_store(response: Response) -> Response:
    # Highscores ändern sich laufend; Browser und Proxies sollen sie nicht zwischenspeichern.
    response.headers["Cache-Control"] = "no-store"
    return response


@bp.post("/runs")
def start_run() -> tuple[Response, int]:
    """Gibt ein signiertes Run-Token für eine neue Spielrunde aus."""
    token = _serializer().dumps({"rid": uuid.uuid4().hex})
    return _no_store(jsonify({"run_token": token})), 201


@bp.get("/highscores")
def list_highscores() -> Response:
    """Liefert die Top-N-Highscores."""
    scores = top_scores(get_db(), current_app.config["HIGHSCORE_TOP_N"])
    return _no_store(jsonify({"highscores": scores}))


@bp.post("/highscores")
def submit_highscore() -> tuple[Response, int]:
    """Nimmt einen neuen Highscore entgegen (Details siehe Moduldokumentation)."""
    config = current_app.config
    db = get_db()
    now = _now()

    # Das Rate-Limit greift vor allem anderen, damit auch fehlerhafte Anfragen zählen.
    allowed, retry_after = register_submission_attempt(
        db,
        _client_hash(),
        now,
        window_s=config["RATE_LIMIT_WINDOW_SECONDS"],
        max_attempts=config["RATE_LIMIT_MAX_SUBMISSIONS"],
    )
    if not allowed:
        response, status = error_response(
            429,
            "rate_limited",
            "Zu viele Einreichungen. Bitte später erneut versuchen.",
            retry_after=retry_after,
        )
        response.headers["Retry-After"] = str(retry_after)
        return response, status

    payload = request.get_json(silent=True)
    if not isinstance(payload, dict):
        return error_response(400, "invalid_json", "Erwartet wird ein JSON-Objekt.")

    rules = PlausibilityRules.from_config(config)
    try:
        submission = parse_submission(payload, rules)
    except SubmissionError as exc:
        return error_response(400, "validation_failed", "Ungültige Eingaben.", fields=exc.fields)

    # Optionale Sperrliste des Betreibers (z. B. Namen realer Personen, Beleidigungen).
    if is_name_blocked(submission.name, load_blocklist(config["NAME_BLOCKLIST_PATH"])):
        return error_response(
            400,
            "validation_failed",
            "Ungültige Eingaben.",
            fields={"name": "Dieser Name ist nicht zulässig."},
        )

    try:
        token_data, issued_at = _serializer().loads(
            submission.run_token,
            max_age=config["RUN_TOKEN_MAX_AGE_SECONDS"],
            return_timestamp=True,
        )
    except SignatureExpired:
        return error_response(422, "run_token_expired", "Das Run-Token ist abgelaufen.")
    except BadSignature:
        return error_response(422, "run_token_invalid", "Das Run-Token ist ungültig.")

    run_id = token_data.get("rid") if isinstance(token_data, dict) else None
    if not isinstance(run_id, str):
        return error_response(422, "run_token_invalid", "Das Run-Token ist ungültig.")

    elapsed_s = (datetime.now(timezone.utc) - issued_at).total_seconds()
    reason = check_plausibility(submission, elapsed_s, rules)
    if reason is not None:
        return error_response(422, "implausible_score", reason)

    # Token einlösen und speichern gehören zusammen: Scheitert das Speichern,
    # darf das Token nicht verbraucht sein.
    with immediate_transaction(db):
        if not consume_run_token(db, run_id, now, config["RUN_TOKEN_MAX_AGE_SECONDS"]):
            return error_response(
                409, "run_token_used", "Für diese Runde wurde bereits ein Highscore gespeichert."
            )
        score_id = insert_score(
            db, submission.name, submission.score, submission.wave, submission.duration_s
        )

    body = {
        "id": score_id,
        "rank": rank_of(db, score_id),
        "name": submission.name,
        "highscores": top_scores(db, config["HIGHSCORE_TOP_N"]),
    }
    return _no_store(jsonify(body)), 201