Personalisierung: Passwort vergessen per Mail-Link, Modus im Check-in, Admin-Reset ohne Änderungszwang (Migration 0067, ADR-0041 18.10.)

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

app.py

@@ -1602,7 +1602,15 @@ def fetch_auftragskatalog_state(device_id: str):
         if key == VEYON_MERKMAL and aktiv:
             optionen = veyon_optionen_anreichern(device_id, merkmal_id, optionen)
         if key == PERSONALISIERUNG_MERKMAL and aktiv:
-            optionen = {**optionen, "anmeldung_anbieten": personalisierung_anmeldung_anbieten(device_id)}
+            modus = personalisierung_modus(device_id)
+            optionen = {
+                **optionen,
+                "modus": modus,
+                # Für Rollenstände vor "modus" (18.10.2026).
+                "anmeldung_anbieten": modus == "personalisieren",
+                "geraet": str(device_id),
+                "kiosk_token": personalisierung_kiosk_token(device_id),
+            }
 
         ergebnis.append({
             "merkmal": key,
@@ -5915,14 +5923,15 @@ def personalisierung_verfallene_markieren(cur) -> None:
 
 
 # Merkmal "Personalisierung möglich": der Check-in sagt der Rolle in den
-# Optionen, ob das Konto "Personalisierung" gerade angeboten wird (ADR-0041,
-# Ergänzung 18.10.2026). Ein persönliches Konto pro Gerät: steht eins und
-# läuft kein Code, verschwindet der Eintrag vom Anmeldebildschirm; ein neuer
-# Code der Verwaltung holt ihn zurück.
+# Optionen, wofür das Konto "Personalisierung" am Anmeldebildschirm gerade da
+# ist (ADR-0041, Ergänzungen 18.10.2026):
+#   personalisieren - kein persönliches Konto oder ein Code läuft
+#   passwort        - ein Konto ist eingerichtet: Eintrag "Passwort vergessen?"
+#   aus             - nur gesperrte Konten: kein Eintrag
 PERSONALISIERUNG_MERKMAL = "personalisierung"
 
 
-def personalisierung_anmeldung_anbieten(device_id: str) -> bool:
+def personalisierung_modus(device_id: str) -> str:
     with get_database_connection() as conn:
         with conn.cursor() as cur:
             personalisierung_verfallene_markieren(cur)
@@ -5930,14 +5939,159 @@ def personalisierung_anmeldung_anbieten(device_id: str) -> bool:
                 """
                 SELECT
                     bool_or(status = ANY(%s)),
-                    bool_or(status = 'angelegt' AND NOT entfernen)
+                    bool_or(status = 'angelegt' AND NOT entfernen),
+                    bool_or(status = 'angelegt' AND NOT entfernen AND NOT gesperrt)
                 FROM personalisierungen
                 WHERE device_id = %s
                 """,
                 (list(PERSONALISIERUNG_LAUFEND), device_id),
             )
-            laufend, konto_vorhanden = cur.fetchone()
-    return bool(laufend) or not konto_vorhanden
+            laufend, konto_vorhanden, konto_aktiv = cur.fetchone()
+    if laufend or not konto_vorhanden:
+        return "personalisieren"
+    return "passwort" if konto_aktiv else "aus"
+
+
+# Der Kiosk "Passwort vergessen?" weist sich mit Gerät und diesem Token aus.
+# Abgeleitet statt gespeichert; lesen kann es jedes lokale Konto des Geräts,
+# es erlaubt aber nur, dem Besitzer einen Link an seine eigene Adresse zu
+# schicken (begrenzt, siehe VERGESSEN_*).
+def personalisierung_kiosk_token(device_id) -> str:
+    schluessel = (ENCRYPTION_KEY or "").encode()
+    if not schluessel:
+        return ""
+    return hmac.new(schluessel, f"personalisierung-kiosk:{device_id}".encode(), hashlib.sha256).hexdigest()[:32]
+
+
+VERGESSEN_LINK_GUELTIG = timedelta(hours=1)
+VERGESSEN_ABSTAND = timedelta(minutes=10)
+VERGESSEN_JE_TAG = 3
+
+
+def email_maskieren(email: str | None) -> str:
+    if not email or "@" not in email:
+        return ""
+    lokal, domain = email.rsplit("@", 1)
+    return f"{lokal[:1]}•••@{domain}"
+
+
+def _vergessen_konto(cur, device_id: str, token: str):
+    """Das aktive persönliche Konto des Geräts, wenn das Kiosk-Token passt."""
+
+    erwartet = personalisierung_kiosk_token(device_id)
+    if not erwartet or not hmac.compare_digest(erwartet, (token or "").strip()):
+        return None
+    cur.execute(
+        """
+        SELECT p.id, p.anmeldename, p.email, p.vergessen_mails, d.hostname, o.name
+        FROM personalisierungen p
+        JOIN devices d ON d.id = p.device_id
+        JOIN organizations o ON o.id = p.organization_id
+        WHERE p.device_id::text = %s AND p.status = 'angelegt' AND NOT p.entfernen AND NOT p.gesperrt
+        ORDER BY p.angelegt_am DESC
+        LIMIT 1
+        """,
+        (str(device_id),),
+    )
+    return cur.fetchone()
+
+
+def vergessen_pruefen(device_id: str, token: str) -> dict | None:
+    with get_database_connection() as conn:
+        with conn.cursor() as cur:
+            row = _vergessen_konto(cur, device_id, token)
+    if row is None:
+        return None
+    _pid, anmeldename, email, _mails, hostname, org_name = row
+    return {"anmeldename": anmeldename, "email_maskiert": email_maskieren(email), "hostname": hostname, "organisation": org_name}
+
+
+def vergessen_anfordern(device_id: str, token: str):
+    """
+    Erzeugt den Link für "Passwort vergessen?". Gibt (Ergebnis, Fehler)
+    zurück; das Ergebnis enthält die volle Adresse und das Token für die
+    Mail, die kundenplattform verschickt.
+    """
+
+    jetzt = datetime.now(timezone.utc)
+    with get_database_connection() as conn:
+        with conn.cursor() as cur:
+            row = _vergessen_konto(cur, device_id, token)
+            if row is None:
+                return None, "Auf diesem Gerät ist kein persönliches Konto eingerichtet."
+            pid, anmeldename, email, mails, hostname, org_name = row
+            cur.execute("SELECT 1 FROM personalisierungen WHERE id = %s FOR UPDATE", (pid,))
+
+            letzte = [m for m in (mails or []) if m > jetzt - timedelta(days=1)]
+            if letzte and max(letzte) > jetzt - VERGESSEN_ABSTAND:
+                return None, "Es wurde gerade erst ein Link verschickt. Bitte im Postfach (auch im Spam-Ordner) nachsehen oder in 10 Minuten erneut versuchen."
+            if len(letzte) >= VERGESSEN_JE_TAG:
+                return None, "Heute wurden schon mehrere Links verschickt. Bitte morgen erneut versuchen oder an die Verwaltung wenden."
+
+            link_token = secrets.token_urlsafe(32)
+            cur.execute(
+                """
+                UPDATE personalisierungen
+                SET vergessen_token_hash = %s, vergessen_bis = %s, vergessen_mails = %s
+                WHERE id = %s
+                """,
+                (personalisierung_token_hash(link_token), jetzt + VERGESSEN_LINK_GUELTIG, letzte + [jetzt], pid),
+            )
+
+    return {
+        "personalisierung_id": str(pid),
+        "token": link_token,
+        "email": email,
+        "email_maskiert": email_maskieren(email),
+        "anmeldename": anmeldename,
+        "hostname": hostname,
+        "organisation": org_name,
+    }, None
+
+
+def _neues_passwort_konto(cur, token: str):
+    cur.execute(
+        """
+        SELECT p.id, p.anmeldename, d.hostname
+        FROM personalisierungen p
+        JOIN devices d ON d.id = p.device_id
+        WHERE p.vergessen_token_hash = %s AND p.vergessen_bis > now()
+          AND p.status = 'angelegt' AND NOT p.entfernen AND NOT p.gesperrt
+        """,
+        (personalisierung_token_hash(token),),
+    )
+    return cur.fetchone()
+
+
+def neues_passwort_pruefen(token: str) -> dict | None:
+    with get_database_connection() as conn:
+        with conn.cursor() as cur:
+            row = _neues_passwort_konto(cur, token)
+    if row is None:
+        return None
+    return {"anmeldename": row[1], "hostname": row[2]}
+
+
+def neues_passwort_setzen(token: str, passwort_hash: str) -> dict | None:
+    """Wie der Reset der Verwaltung, nur ohne Änderungszwang; der Link ist danach verbraucht."""
+
+    with get_database_connection() as conn:
+        with conn.cursor() as cur:
+            row = _neues_passwort_konto(cur, token)
+            if row is None:
+                return None
+            pid, anmeldename, hostname = row
+            cur.execute(
+                """
+                UPDATE personalisierungen
+                SET reset_id = %s, reset_passwort_hash = %s, reset_aendern = FALSE,
+                    reset_angefordert_am = now(), reset_angefordert_von = 'selbst (Mail-Link)',
+                    reset_gesetzt_am = NULL, vergessen_token_hash = NULL, vergessen_bis = NULL
+                WHERE id = %s
+                """,
+                (uuid4(), passwort_hash, pid),
+            )
+    return {"anmeldename": anmeldename, "hostname": hostname}
 
 
 PERSONALISIERUNG_SPALTEN = (
@@ -6368,7 +6522,7 @@ class PersoenlichesKontoErgebnis(BaseModel):
 
 class PasswortResetRequest(BaseModel):
     passwort_hash: str
-    aendern: bool = True
+    aendern: bool = False
     angefordert_von: str | None = None
 
 
@@ -6397,6 +6551,16 @@ class ConfirmPersonalisierungRequest(BaseModel):
     token: str
 
 
+class VergessenRequest(BaseModel):
+    device_id: str
+    token: str
+
+
+class NeuesPasswortRequest(BaseModel):
+    token: str
+    passwort_hash: str | None = None
+
+
 class SetEmailDomaenenRequest(BaseModel):
     domaenen: list[str]
 
@@ -6537,6 +6701,46 @@ def post_personalisierung_einreichen(
     return {"success": True, **ergebnis}
 
 
+@app.post("/api/v1/personalisierung/vergessen/pruefen")
+def post_vergessen_pruefen(payload: VergessenRequest, authorization: str | None = Header(default=None)):
+    if not require_service_token(authorization):
+        return {"success": False, "error": "unauthorized", "message": "Fehlendes oder ungültiges Service-Token."}
+
+    ergebnis = vergessen_pruefen(payload.device_id, payload.token)
+    if ergebnis is None:
+        return {"success": False, "error": "kein_konto", "message": "Auf diesem Gerät ist kein persönliches Konto eingerichtet."}
+    return {"success": True, **ergebnis}
+
+
+@app.post("/api/v1/personalisierung/vergessen/anfordern")
+def post_vergessen_anfordern(payload: VergessenRequest, authorization: str | None = Header(default=None)):
+    if not require_service_token(authorization):
+        return {"success": False, "error": "unauthorized", "message": "Fehlendes oder ungültiges Service-Token."}
+
+    ergebnis, fehler = vergessen_anfordern(payload.device_id, payload.token)
+    if ergebnis is None:
+        return {"success": False, "error": "abgelehnt", "message": fehler}
+    return {"success": True, **ergebnis}
+
+
+@app.post("/api/v1/personalisierung/neues-passwort")
+def post_neues_passwort(payload: NeuesPasswortRequest, authorization: str | None = Header(default=None)):
+    """Ohne passwort_hash nur prüfen (Seite hinter dem Link), mit Hash setzen."""
+
+    if not require_service_token(authorization):
+        return {"success": False, "error": "unauthorized", "message": "Fehlendes oder ungültiges Service-Token."}
+
+    if payload.passwort_hash is None:
+        ergebnis = neues_passwort_pruefen(payload.token)
+    elif not payload.passwort_hash.startswith("$6$"):
+        return {"success": False, "error": "invalid_request", "message": "Kein SHA-512-crypt-Hash."}
+    else:
+        ergebnis = neues_passwort_setzen(payload.token, payload.passwort_hash)
+    if ergebnis is None:
+        return {"success": False, "error": "ungueltig", "message": "Der Link ist ungültig, schon verwendet oder abgelaufen."}
+    return {"success": True, **ergebnis}
+
+
 @app.post("/api/v1/personalisierung/bestaetigen")
 def post_personalisierung_bestaetigen(
     payload: ConfirmPersonalisierungRequest,