3d.lehrstun.de/services/auth.py

"""Login-/Auth-Helper.

Konventionen:
- PIN-Hashing: bcrypt
- Persistenter Login-Lockout in DB-Tabelle login_lockout (V7, siehe decisions.md#E-020)
- Decorators: @login_required (Profil-Session), @admin_required (Admin-Session),
  @onboarding_l1_fertig (V2)
"""
import os
from datetime import datetime, timedelta
from functools import wraps

import bcrypt
from flask import session, redirect, url_for, flash, request

from database import get_db

# Lockout-Konfiguration
LOCKOUT_FENSTER_MINUTEN = 5      # Fehlversuche werden innerhalb dieses Fensters gezaehlt
LOCKOUT_MAX_VERSUCHE = 5         # Nach so vielen Fehlversuchen: Sperre
LOCKOUT_DAUER_MINUTEN = 15       # So lange wird gesperrt
ONBOARDING_L1 = 1                # Lektions-Nummer von L1 (Sicherheits-Lektion)


# ===========================================================================
# PIN-Handling
# ===========================================================================

def pin_hashen(pin: str) -> str:
    """Hasht eine PIN mit bcrypt. Default cost ist OK fuer 4-stellige PINs."""
    return bcrypt.hashpw(pin.encode("utf-8"), bcrypt.gensalt()).decode("utf-8")


def pin_pruefen(pin: str, hash_str: str) -> bool:
    """Prueft eine PIN gegen den gespeicherten Hash."""
    try:
        return bcrypt.checkpw(pin.encode("utf-8"), hash_str.encode("utf-8"))
    except (ValueError, TypeError):
        return False


# ===========================================================================
# Persistenter Login-Lockout (V7, verbessert ggue Arduino-Kurs)
# ===========================================================================

def ist_gesperrt(profil_id: int) -> tuple[bool, int]:
    """Liefert (gesperrt: bool, sek_rest: int)."""
    conn = get_db()
    row = conn.execute(
        "SELECT sperre_bis FROM login_lockout WHERE profil_id = ? "
        "ORDER BY id DESC LIMIT 1",
        (profil_id,)
    ).fetchone()
    conn.close()
    if not row or not row["sperre_bis"]:
        return (False, 0)
    try:
        sperre_bis = datetime.fromisoformat(row["sperre_bis"])
    except (ValueError, TypeError):
        return (False, 0)
    jetzt = datetime.now()
    if sperre_bis <= jetzt:
        return (False, 0)
    return (True, int((sperre_bis - jetzt).total_seconds()))


def fehlversuch_loggen(profil_id: int) -> None:
    """Loggt einen Fehlversuch. Bei zu vielen innerhalb des Fensters: setzt Sperre."""
    conn = get_db()
    jetzt = datetime.now()
    fenster_start = (jetzt - timedelta(minutes=LOCKOUT_FENSTER_MINUTEN)).isoformat()

    # Zaehle Fehlversuche im Fenster (ungelaufene, also ohne aktive Sperre)
    versuche_im_fenster = conn.execute(
        "SELECT COUNT(*) AS n FROM login_lockout "
        "WHERE profil_id = ? AND letzter_versuch_am > ?",
        (profil_id, fenster_start)
    ).fetchone()["n"]

    sperre_bis = None
    if versuche_im_fenster + 1 >= LOCKOUT_MAX_VERSUCHE:
        sperre_bis = (jetzt + timedelta(minutes=LOCKOUT_DAUER_MINUTEN)).isoformat()

    conn.execute(
        "INSERT INTO login_lockout (profil_id, versuche_count, sperre_bis, letzter_versuch_am) "
        "VALUES (?, ?, ?, ?)",
        (profil_id, versuche_im_fenster + 1, sperre_bis, jetzt.isoformat())
    )
    conn.commit()
    conn.close()


def lockout_zuruecksetzen(profil_id: int) -> None:
    """Bei erfolgreichem Login: alte Lockout-Eintraege loeschen."""
    conn = get_db()
    conn.execute("DELETE FROM login_lockout WHERE profil_id = ?", (profil_id,))
    conn.commit()
    conn.close()


# ===========================================================================
# Decorators
# ===========================================================================

def login_required(f):
    """Profil muss eingeloggt sein, sonst Redirect zum Login."""
    @wraps(f)
    def wrapper(*args, **kwargs):
        if not session.get("profil_id"):
            flash("Bitte erst einloggen.", "error")
            return redirect(url_for("profil.login"))
        return f(*args, **kwargs)
    return wrapper


def admin_required(f):
    """Admin-Session noetig."""
    @wraps(f)
    def wrapper(*args, **kwargs):
        if not session.get("admin"):
            return redirect(url_for("admin.admin_login"))
        return f(*args, **kwargs)
    return wrapper


def onboarding_l1_fertig(f):
    """V2: blockiert Slot-Routes, bis L1 (Sicherheits-Lektion) abgeschlossen ist."""
    @wraps(f)
    def wrapper(*args, **kwargs):
        if not session.get("profil_id"):
            flash("Bitte erst einloggen.", "error")
            return redirect(url_for("profil.login"))
        conn = get_db()
        row = conn.execute(
            "SELECT status FROM profil_lektion_status "
            "WHERE profil_id = ? AND lektion_nummer = ?",
            (session["profil_id"], ONBOARDING_L1)
        ).fetchone()
        conn.close()
        if not row or row["status"] != "fertig":
            flash(
                "Du musst erst die Onboarding-Lektion L1 (Sicherheit) abschliessen, "
                "bevor du Drucker reservieren darfst.",
                "error"
            )
            return redirect(url_for("profil.cockpit"))
        return f(*args, **kwargs)
    return wrapper


def admin_passwort_pruefen(eingegeben: str) -> bool:
    """Prueft Admin-Passwort aus .env."""
    erwartet = os.environ.get("ADMIN_PASSWORT", "")
    if not erwartet:
        return False
    return eingegeben == erwartet