tesm-license/apps/tesm-license/src/tesm_license/blueprints/selfservice.py

"""Self-Service je Ticket -- ohne Anmeldung, allein ueber den Ticket-Link.

Die Berechtigung ist der Besitz der Ticket-ID: 192 Bit Zufall
(``secrets.token_urlsafe(24)``), zugleich Primaerschluessel. Damit kann ein Kunde
seine Lizenz herunterladen, bei einem Systemwechsel freigeben und neu ausstellen
lassen -- ohne Support-Anruf.

Fuer Tickets mit einer anderen Rolle als ``customer`` antworten **alle** Routen
mit 404, nicht mit "keine Berechtigung": aus der Fehlermeldung soll sich nicht
ablesen lassen, dass eine geratene ID zu einem Lizenzserver-Ticket gehoert.
"""

from __future__ import annotations

from flask import Blueprint, Response, abort, flash, redirect, render_template, request, url_for
from tesm_core import audit
from tesm_core.auth.service import client_ip
from tesm_core.extension import core, db, transaction
from tesm_core.security import ratelimit

import tesm_licensing as lic

from ..services import activation, issuance, reporting
from .common import (
    bundle_bytes,
    customer_facing_url,
    license_filename,
    master_endpoint,
    signing_key,
    type_labels,
    vendor_info,
)

bp = Blueprint("selfservice", __name__, url_prefix="/lizenz")

RATE_BUCKET = "selfservice"
RATE_LIMIT = 20
RATE_WINDOW_MINUTES = 10


def _ticket_or_404(ticket_id: str) -> dict:
    conn = db()
    if not core().settings.get(conn, "self_service_enabled"):
        abort(404)
    ticket = issuance.get_ticket(conn, ticket_id)
    if ticket is None or str(ticket["role"]) != "customer":
        abort(404)
    return ticket


def _rate_guard() -> None:
    conn = db()
    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:
        abort(429, description="Zu viele Anfragen. Bitte spaeter erneut versuchen.")


def _needs_offline(conn, ticket_id: str):  # type: ignore[no-untyped-def]
    """Die Lizenz, fuer die der Offline-Weg gerade einen Zweck hat -- sonst ``None``.

    Zwei Faelle, und nur diese:

    * **Noch nicht aktiviert.** Der Kunde muss aktivieren.
    * **Neue Revision ausstehend.** Der Anbieter hat nachsigniert (andere
      Module, laengere Laufzeit); eine Installation ohne Verbindung holt sie
      sonst nie ab.

    Ist die Lizenz aktiv und auf dem aktuellen Stand, gibt es nichts zu tun --
    fuer einen Systemwechsel gibt es die Freigabe.
    """
    for row in issuance.list_licenses(conn, ticket_id=ticket_id):
        status = str(row["status"])
        if status == "issued":
            return row
        if status == "active" and issuance.revision_pending(row):
            return row
    return None


def _log(action: str, ticket: dict, detail: str = "", severity: str = "notice") -> None:
    extension = core()
    with transaction() as conn:
        audit.log(
            conn,
            actor="self-service",
            action=action,
            target=str(ticket["customer_name"]),
            detail=detail,
            ip=client_ip(),
            severity=severity,
        )
    live = extension.registry.get("live_log")
    if live is not None:
        live.info("selfservice", f"{action} {ticket['customer_name']} {detail}".strip())


@bp.get("/<ticket_id>")
def overview(ticket_id: str):
    _rate_guard()
    ticket = _ticket_or_404(ticket_id)
    conn = db()
    licenses = [
        {**row, **issuance.display_state(row)}
        for row in issuance.list_licenses(conn, ticket_id=ticket_id)
    ]
    current = next((row for row in licenses if row["status"] in {"issued", "active"}), None)
    return render_template(
        "license/selfservice.html",
        offline_needed=_needs_offline(db(), ticket_id),
        ticket=ticket,
        current=current,
        history=[row for row in licenses if row is not current],
        type_labels=type_labels(),
        vendor=vendor_info(),
        portal_url=url_for("portal.request_link"),
        allow_revoke=bool(core().settings.get(conn, "allow_customer_revoke")),
        blocked=str(ticket["status"]) == "blocked",
    )


@bp.post("/<ticket_id>/ausstellen")
def issue(ticket_id: str):
    _rate_guard()
    ticket = _ticket_or_404(ticket_id)
    conn = db()
    if str(ticket["status"]) == "blocked":
        flash(
            "Dieses Ticket ist gesperrt. Bitte wenden Sie sich an Ihren Ansprechpartner.",
            "danger",
        )
        return redirect(url_for("selfservice.overview", ticket_id=ticket_id))
    if issuance.ticket_has_open_license(conn, ticket_id):
        flash(
            "Es existiert bereits eine gueltige Lizenz. Geben Sie diese zuerst frei, wenn Sie "
            "auf ein anderes System wechseln moechten.",
            "warning",
        )
        return redirect(url_for("selfservice.overview", ticket_id=ticket_id))

    key = signing_key()
    try:
        with transaction():
            result = issuance.issue_from_ticket(
                conn,
                ticket_id,
                master_private_key=key.private_key,
                master_public_key=key.public_key,
                master_endpoint=master_endpoint(),
                vendor=vendor_info(),
                actor="self-service",
                keystore=core().keystore,
            )
    except issuance.IssuanceError as exc:
        flash(str(exc), "danger")
        return redirect(url_for("selfservice.overview", ticket_id=ticket_id))

    _log("license.issued_self_service", ticket, result.license_id[:12])
    reporting.report_soon(core())
    flash(
        "Ihre Lizenz wurde erstellt. Laden Sie die Datei herunter und spielen Sie sie in Ihrer "
        "TESM-Installation ein.",
        "success",
    )
    return redirect(url_for("selfservice.overview", ticket_id=ticket_id))


@bp.get("/<ticket_id>/datei/<license_id>")
def download(ticket_id: str, license_id: str):
    _rate_guard()
    ticket = _ticket_or_404(ticket_id)
    conn = db()
    row = issuance.get_license(conn, license_id)
    if row is None or str(row["ticket_id"]) != ticket_id:
        abort(404)
    if str(row["status"]) == "revoked":
        abort(410, description="Diese Lizenz wurde widerrufen.")
    try:
        bundle = issuance.delivery_bundle(conn, license_id, keystore=core().keystore)
    except issuance.IssuanceError as exc:
        abort(409, description=str(exc))

    with transaction():
        issuance.record_delivery(
            conn,
            license_id=license_id,
            channel="self-service",
            recipient=str(ticket["contact_email"] or ""),
            actor="self-service",
        )
    _log("license.downloaded_self_service", ticket, license_id[:12], severity="info")
    return Response(
        bundle_bytes(bundle),
        mimetype="application/json",
        headers={
            "Content-Disposition": (
                f'attachment; filename="{license_filename(license_id, str(ticket["customer_name"]))}"'
            ),
            "Cache-Control": "no-store",
        },
    )


@bp.route("/<ticket_id>/offline", methods=["GET", "POST"])
def offline(ticket_id: str):
    """Loest einen Aktivierungscode ein, wenn TESM den Server nicht erreicht."""
    ticket = _ticket_or_404(ticket_id)
    conn = db()
    pending = _needs_offline(conn, ticket_id)
    if pending is None:
        # Ohne noch nicht aktivierte Lizenz gibt es hier nichts zu tun.
        return redirect(url_for("selfservice.overview", ticket_id=ticket_id))

    response_code = ""
    summary: dict[str, object] = {}

    if request.method == "POST":
        _rate_guard()
        raw = request.form.get("code", "")
        try:
            payload = lic.decode_code(raw)
        except ValueError as exc:
            flash(f"Der Code ist unlesbar: {exc}", "danger")
            return redirect(url_for("selfservice.offline", ticket_id=ticket_id))

        # Nur Lizenzen dieses Vorgangs -- ein fremder Code gehoert nicht hierher.
        row = issuance.get_license(conn, str(payload.get("license_id") or ""))
        if row is None or str(row["ticket_id"]) != ticket_id:
            _log(
                "selfservice.offline_rejected",
                ticket,
                "Code gehoert nicht zu diesem Vorgang",
                severity="warning",
            )
            flash(
                "Dieser Code gehoert nicht zu Ihrem Vorgang. Bitte pruefen Sie, ob Sie ihn "
                "vollstaendig kopiert haben.",
                "danger",
            )
            return redirect(url_for("selfservice.offline", ticket_id=ticket_id))

        key = signing_key()
        with transaction():
            outcome = activation.process(conn, payload, master_private_key=key.private_key)
        summary = {"action": payload.get("action"), "hostname": payload.get("hostname")}
        if outcome.ok:
            response_code = lic.encode_code(outcome.response)
            _log(
                f"selfservice.offline_{payload.get('action')}",
                ticket,
                f"Lizenz {str(payload.get('license_id'))[:12]}",
            )
        else:
            message = str(outcome.response.get("message") or "Die Anfrage wurde abgewiesen.")
            _log("selfservice.offline_rejected", ticket, message, severity="warning")
            flash(message, "danger")

    return render_template(
        "license/selfservice_offline.html",
        ticket=ticket,
        license=pending,
        response_code=response_code,
        summary=summary,
        vendor=vendor_info(),
    )


@bp.post("/<ticket_id>/freigeben/<license_id>")
def revoke(ticket_id: str, license_id: str):
    """Gibt die aktuelle Lizenz frei -- fuer einen Systemwechsel.

    Fachlich ein Widerruf: die Datei gilt nicht mehr, das Ticket bleibt offen,
    eine neue Lizenz kann sofort ausgestellt werden.
    """
    _rate_guard()
    ticket = _ticket_or_404(ticket_id)
    conn = db()
    if not core().settings.get(conn, "allow_customer_revoke"):
        abort(404)
    row = issuance.get_license(conn, license_id)
    if row is None or str(row["ticket_id"]) != ticket_id:
        abort(404)
    try:
        with transaction():
            issuance.revoke_license(conn, license_id, reason="Freigabe durch den Kunden")
    except issuance.IssuanceError as exc:
        flash(str(exc), "danger")
        return redirect(url_for("selfservice.overview", ticket_id=ticket_id))

    _log("license.revoked_self_service", ticket, license_id[:12], severity="warning")
    reporting.report_soon(core())
    flash(
        "Die Lizenz wurde freigegeben. Sie koennen jetzt eine neue Lizenz fuer Ihr neues System "
        "ausstellen.",
        "success",
    )
    return redirect(url_for("selfservice.overview", ticket_id=ticket_id))