"""Kundenportal -- zeitlich begrenzter Zugang ohne Konto. Zweistufig, bewusst: 1. ``/portal`` nimmt eine E-Mail-Adresse an und verschickt einen **einmal** verwendbaren Zugangslink mit kurzer Gueltigkeit. 2. Der Link erzeugt eine Sitzung mit eigener, unabhaengiger Gueltigkeit. Der Vorgaenger hatte zuerst einen permanenten, nie ablaufenden Link pro Kunde (``/portal/``). Das wurde als Schwachstelle erkannt und ersetzt -- hier existiert die permanente Variante gar nicht. Anti-Enumeration: die Antwort ist immer dieselbe, unabhaengig davon, ob die Adresse bekannt ist. Ein Fehler beim Mailversand wird dem Anfragenden gegenueber verschwiegen, aber serverseitig protokolliert -- sonst waere eine kaputte Mailkonfiguration unsichtbar. """ from __future__ import annotations import logging import secrets from datetime import datetime, timedelta, timezone from typing import Any from flask import Blueprint, flash, redirect, render_template, request, session, url_for from tesm_core import audit from tesm_core.auth.service import client_ip from tesm_core.db import Database from tesm_core.extension import core, db, transaction from tesm_core.security import ratelimit from ..services import issuance, mail from .common import customer_facing_url, vendor_info logger = logging.getLogger(__name__) bp = Blueprint("portal", __name__, url_prefix="/portal") SESSION_CUSTOMER = "portal_customer_id" SESSION_EXPIRES = "portal_expires_at" RATE_BUCKET = "portal:request" RATE_LIMIT = 5 RATE_WINDOW_MINUTES = 15 GENERIC_ANSWER = ( "Falls diese E-Mail-Adresse bei uns hinterlegt ist, wurde soeben eine Nachricht mit " "einem {minutes} Minuten gültigen Zugangslink verschickt." ) EDITABLE_FIELDS = ( "contact_person", "contact_phone", "contact_email", "street", "house_number", "postal_code", "city", ) def _now() -> datetime: return datetime.now(timezone.utc) def _iso(value: datetime) -> str: return value.isoformat(timespec="seconds") def _parse(value: Any) -> datetime | None: """Liest einen Zeitstempel aus der Datenbank. Bewusst geparst statt als Zeichenkette verglichen: ein Wert, der ueber ``datetime('now')`` von SQLite selbst geschrieben wurde, hat ein Leerzeichen statt ``T`` und keine Zeitzone -- ein lexikografischer Vergleich mit unserem ISO-Format waere dann falsch, und zwar zugunsten von "abgelaufen". """ if not value: return None text = str(value).strip().replace(" ", "T") if text.endswith("Z"): text = text[:-1] + "+00:00" try: parsed = datetime.fromisoformat(text) except ValueError: return None return parsed if parsed.tzinfo else parsed.replace(tzinfo=timezone.utc) def _enabled() -> bool: return bool(core().settings.get(db(), "portal_enabled")) @bp.route("/", methods=["GET", "POST"]) def request_link(): extension = core() conn = db() if not _enabled(): return render_template("license/portal_disabled.html"), 404 minutes = int(extension.settings.get(conn, "portal_link_minutes") or 30) if request.method == "POST": email = (request.form.get("email") or "").strip() verdict = ratelimit.check( conn, RATE_BUCKET, client_ip() or "unbekannt", limit=RATE_LIMIT, window_minutes=RATE_WINDOW_MINUTES, ) with transaction(): ratelimit.record(conn, RATE_BUCKET, client_ip() or "unbekannt") if verdict.blocked: # Auch hier keine abweichende Aussage -- nur eine Bremse. flash(GENERIC_ANSWER.format(minutes=minutes), "info") return redirect(url_for("portal.request_link")) customer = issuance.find_customer_by_email(conn, email) if customer is not None: token = secrets.token_urlsafe(32) with transaction(): conn.execute( "INSERT INTO portal_access_tokens (token, customer_id, created_at, " "expires_at, created_ip) VALUES (?,?,?,?,?)", ( token, int(customer["id"]), _iso(_now()), _iso(_now() + timedelta(minutes=minutes)), client_ip() or "", ), ) audit.log( conn, actor="portal", action="portal.link_requested", target=str(customer["name"]), ip=client_ip(), severity="notice", ) access_url = customer_facing_url("portal.access", token=token) vendor = vendor_info() try: mail.send( conn, to=str(customer["contact_email"]), subject="Ihr Zugang zum Kundenportal", body="\n".join( [ f"Guten Tag{(' ' + str(customer['contact_person'])) if customer['contact_person'] else ''},", "", "mit dem folgenden Link erreichen Sie Ihre Daten im Kundenportal.", f"Der Link ist {minutes} Minuten gültig und lässt sich nur einmal verwenden:", "", access_url, "", "Wenn Sie diesen Zugang nicht angefragt haben, ignorieren Sie die Nachricht.", "", "Freundliche Grüße", vendor.get("name") or "Ihr Lizenzgeber", ] ), ) except mail.MailError as exc: # Bewusst nicht nach aussen tragen (Anti-Enumeration), aber # serverseitig sichtbar machen. logger.error("Portal-Mail an %s fehlgeschlagen: %s", email, exc) flash(GENERIC_ANSWER.format(minutes=minutes), "info") return redirect(url_for("portal.request_link")) return render_template("license/portal_request.html", minutes=minutes) @bp.get("/zugang/") def access(token: str): extension = core() conn = db() if not _enabled(): return render_template("license/portal_disabled.html"), 404 row = Database.one( conn, "SELECT * FROM portal_access_tokens WHERE token=?", (token,) ) expires = _parse(row["expires_at"]) if row is not None else None valid = row is not None and row["used_at"] is None and expires is not None and _now() < expires if not valid: flash("Dieser Link ist ungültig oder abgelaufen.", "danger") return redirect(url_for("portal.request_link")) minutes = int(extension.settings.get(conn, "portal_session_minutes") or 60) with transaction(): # Einmalverwendung erzwingen: auch innerhalb der Gueltigkeit scheitert # ein zweiter Aufruf. conn.execute( "UPDATE portal_access_tokens SET used_at=? WHERE token=? AND used_at IS NULL", (_iso(_now()), token), ) audit.log( conn, actor="portal", action="portal.session_started", target=str(row["customer_id"]), ip=client_ip(), severity="notice", ) session[SESSION_CUSTOMER] = int(row["customer_id"]) session[SESSION_EXPIRES] = _iso(_now() + timedelta(minutes=minutes)) return redirect(url_for("portal.me")) def _current_customer() -> dict[str, Any] | None: customer_id = session.get(SESSION_CUSTOMER) expires = _parse(session.get(SESSION_EXPIRES)) if not customer_id or expires is None or _now() >= expires: session.pop(SESSION_CUSTOMER, None) session.pop(SESSION_EXPIRES, None) return None return issuance.get_customer(db(), int(customer_id)) @bp.route("/meine-daten", methods=["GET", "POST"]) def me(): conn = db() if not _enabled(): return render_template("license/portal_disabled.html"), 404 customer = _current_customer() if customer is None: flash("Ihre Sitzung ist abgelaufen. Bitte fordern Sie einen neuen Zugangslink an.", "info") return redirect(url_for("portal.request_link")) if request.method == "POST": fields = {key: request.form.get(key, "") for key in EDITABLE_FIELDS} try: with transaction(): changes = issuance.update_customer( conn, int(customer["id"]), name=str(customer["name"]), **fields ) if changes: audit.log( conn, actor="portal", action="portal.customer_updated", target=str(customer["name"]), detail=", ".join(changes), ip=client_ip(), severity="notice", ) flash("Ihre Angaben wurden gespeichert." if changes else "Keine Änderungen.", "success") except issuance.IssuanceError as exc: flash(str(exc), "danger") return redirect(url_for("portal.me")) tickets = issuance.list_tickets(conn, customer_id=int(customer["id"])) licenses = { str(row["ticket_id"]): {**row, **issuance.display_state(row)} for row in issuance.list_licenses(conn, customer_id=int(customer["id"])) if row["status"] in {"issued", "active"} } self_service_enabled = bool(core().settings.get(conn, "self_service_enabled")) return render_template( "license/portal_me.html", customer=customer, tickets=tickets, licenses=licenses, expires_at=session.get(SESSION_EXPIRES), self_service_enabled=self_service_enabled, vendor=vendor_info(), ) @bp.post("/abmelden") def logout(): session.pop(SESSION_CUSTOMER, None) session.pop(SESSION_EXPIRES, None) flash("Sie wurden abgemeldet.", "info") return redirect(url_for("portal.request_link"))