diff --git a/apps/tesm/pyproject.toml b/apps/tesm/pyproject.toml
index 42aa607..a54c3b4 100644
--- a/apps/tesm/pyproject.toml
+++ b/apps/tesm/pyproject.toml
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project]
name = "tesm"
-version = "2.0.2"
+version = "2.0.3"
description = "TESM -- Verwaltung und Ueberwachung von PoE-Endgeraeten und Aruba-Switchen"
requires-python = ">=3.11"
dependencies = [
diff --git a/apps/tesm/src/tesm/cli.py b/apps/tesm/src/tesm/cli.py
index 73911bc..dc88ae1 100644
--- a/apps/tesm/src/tesm/cli.py
+++ b/apps/tesm/src/tesm/cli.py
@@ -243,6 +243,131 @@ def cmd_convert_legacy(args: argparse.Namespace) -> int:
return 0
+
+def cmd_legacy_export(args: argparse.Namespace) -> int:
+ """Liest eine alte Installation und schreibt deren Daten als Exportdatei.
+
+ Der Vorgaenger kann nur ueber die Weboberflaeche exportieren: anmelden,
+ Kategorien anhaken, Datei herunterladen. Fuer ein Upgrade ist das der
+ falsche Weg -- es soll ohne Browser laufen und auch dann noch, wenn die
+ alte Anwendung nicht mehr startet. Gelesen wird ausschliesslich; die alte
+ Installation bleibt unberuehrt.
+
+ Herauskommt eine Datei im **alten** Format. Damit ist sie genau das, was
+ der Vorgaenger selbst exportiert haette -- und taugt deshalb auch als
+ Rueckweg.
+ """
+ from getpass import getpass
+ from pathlib import Path
+
+ from .services import legacy_transfer as legacy
+
+ quelle = Path(args.source)
+ ziel = Path(args.out)
+ if ziel.exists() and not args.force:
+ print(f"{ziel} existiert bereits. Mit --force ueberschreiben.")
+ return 2
+
+ passphrase = args.passphrase or getpass("Passphrase fuer die Exportdatei: ")
+ try:
+ nutzlast, bericht = legacy.read_installation(quelle)
+ inhalt = legacy.seal_legacy(nutzlast, passphrase)
+ except legacy.LegacyTransferError as exc:
+ print(f"Fehlgeschlagen: {exc}", file=sys.stderr)
+ return 1
+
+ print(bericht.as_text("Gelesen"))
+ if args.dry_run:
+ print()
+ print("--dry-run: keine Datei geschrieben.")
+ return 0
+ try:
+ ziel.write_text(inhalt, encoding="utf-8", newline="\n")
+ except OSError as exc:
+ print(f"Schreiben fehlgeschlagen: {exc}", file=sys.stderr)
+ return 1
+ print()
+ print(f"Geschrieben: {ziel}")
+ return 0
+
+
+def cmd_legacy_import(args: argparse.Namespace) -> int:
+ """Liest eine Datei im neuen Format ein -- derselbe Weg wie in der Oberflaeche.
+
+ Gebraucht wird das fuer das Upgrade: dort laeuft niemand durch einen
+ Browser. Geschrieben wird ueber ``transfer.apply_import`` -- dieselbe
+ Funktion, die die Seite "Sicherung" benutzt. Ein zweiter Schreibpfad fuer
+ Fremddaten waere die Stelle, die niemand pflegt.
+ """
+ from getpass import getpass
+ from pathlib import Path
+
+ from .services import transfer
+
+ quelle = Path(args.datei)
+ if not quelle.is_file():
+ print(f"Datei nicht gefunden: {quelle}", file=sys.stderr)
+ return 2
+
+ app, extension = _app_and_core()
+ passphrase = args.passphrase or getpass("Passphrase der Datei: ")
+ try:
+ nutzlast, kopf = transfer.unseal(quelle.read_bytes(), passphrase)
+ except transfer.TransferError as exc:
+ print(f"Fehlgeschlagen: {exc}", file=sys.stderr)
+ return 1
+
+ vorhanden = [eintrag["key"] for eintrag in transfer.summarize(nutzlast)]
+ gewaehlt = (
+ [key for key in args.categories.split(",") if key.strip()]
+ if args.categories
+ else vorhanden
+ )
+ unbekannt = sorted(set(gewaehlt) - set(transfer.CATEGORY_BY_KEY))
+ if unbekannt:
+ print(f"Unbekannte Kategorie(n): {', '.join(unbekannt)}", file=sys.stderr)
+ return 2
+
+ print(f"Datei vom {kopf.get('created_at') or 'unbekannt'}:")
+ for eintrag in transfer.summarize(nutzlast):
+ marke = "x" if eintrag["key"] in gewaehlt else " "
+ print(f" [{marke}] {eintrag['label']}: {eintrag['count']}")
+ if args.dry_run:
+ print()
+ print("--dry-run: nichts geschrieben.")
+ return 0
+
+ with app.app_context(), extension.database.session() as conn:
+ with extension.database.transaction(conn):
+ bericht = transfer.apply_import(
+ conn,
+ extension.keystore,
+ nutzlast,
+ categories=gewaehlt,
+ set_setting=lambda key, value: extension.settings.set(
+ conn, key, value, actor=args.actor
+ ),
+ )
+ audit.log(
+ conn,
+ actor=args.actor,
+ action="backup.imported",
+ detail="; ".join(
+ f"{key}: {zahlen['created']} neu, {zahlen['updated']} aktualisiert, "
+ f"{zahlen['skipped']} uebersprungen"
+ for key, zahlen in bericht.items()
+ )
+ + " (Kommandozeile)",
+ severity="warning",
+ )
+ print()
+ for key, zahlen in sorted(bericht.items()):
+ print(
+ f" {key}: {zahlen['created']} neu, {zahlen['updated']} aktualisiert, "
+ f"{zahlen['skipped']} uebersprungen"
+ )
+ return 0
+
def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(
prog="tesm-admin", description="Verwaltungswerkzeuge fuer TESM"
@@ -288,6 +413,46 @@ def build_parser() -> argparse.ArgumentParser:
legacy.add_argument("--force", action="store_true", help="Zieldatei ueberschreiben")
legacy.set_defaults(func=cmd_convert_legacy)
+
+ export_alt = sub.add_parser(
+ "legacy-export",
+ help="Daten einer alten Installation als Exportdatei sichern",
+ description=(
+ "Liest die Datenbank einer Installation der Vorgaengerfassung und schreibt "
+ "Clients, Switche, Zugangsdaten und DHCP in eine Datei im alten Format. "
+ "Die alte Installation wird nur gelesen. Ohne Browser und auch dann noch "
+ "moeglich, wenn die alte Anwendung nicht mehr startet."
+ ),
+ )
+ export_alt.add_argument(
+ "--source", default="/srv/tesm", help="Verzeichnis der alten Installation"
+ )
+ export_alt.add_argument("--out", required=True, help="Zieldatei")
+ export_alt.add_argument("--passphrase", help="Passphrase der Zieldatei (sonst wird gefragt)")
+ export_alt.add_argument("--dry-run", action="store_true", help="Nur den Bericht zeigen")
+ export_alt.add_argument("--force", action="store_true", help="Zieldatei ueberschreiben")
+ export_alt.set_defaults(func=cmd_legacy_export)
+
+ import_neu = sub.add_parser(
+ "legacy-import",
+ help="Datei im Format dieser Anwendung einlesen",
+ description=(
+ "Liest eine Sicherungs- oder uebersetzte Datei ein -- derselbe Weg wie unter "
+ "Verwaltung -> Sicherung, nur ohne Browser. Ohne --categories werden alle "
+ "Kategorien uebernommen, die in der Datei stehen."
+ ),
+ )
+ import_neu.add_argument("datei", help="Datei im Format dieser Anwendung")
+ import_neu.add_argument("--passphrase", help="Passphrase der Datei (sonst wird gefragt)")
+ import_neu.add_argument(
+ "--categories", help="Kommaliste, z.B. devices,switches,credentials,dhcp"
+ )
+ import_neu.add_argument(
+ "--actor", default="upgrade", help="Name, der im Auditlog erscheint"
+ )
+ import_neu.add_argument("--dry-run", action="store_true", help="Nur zeigen, was drin steht")
+ import_neu.set_defaults(func=cmd_legacy_import)
+
add_web_setup(sub, cmd_web_setup)
return parser
diff --git a/apps/tesm/src/tesm/services/legacy_transfer.py b/apps/tesm/src/tesm/services/legacy_transfer.py
index a990eda..f9a579c 100644
--- a/apps/tesm/src/tesm/services/legacy_transfer.py
+++ b/apps/tesm/src/tesm/services/legacy_transfer.py
@@ -27,6 +27,7 @@ from __future__ import annotations
import base64
import json
from dataclasses import dataclass, field
+from pathlib import Path
from typing import Any
#: Runden der alten Schluesselableitung. Steht so im Vorgaenger und darf sich
@@ -84,8 +85,13 @@ class Report:
if text not in self.notes:
self.notes.append(text)
- def as_text(self) -> str:
- zeilen = ["Uebersetzt:"]
+ def as_text(self, titel: str = "Uebersetzt") -> str:
+ """Bericht als Text. Der Titel gehoert dem Aufrufer.
+
+ Beim Lesen einer alten Installation ist noch nichts uebersetzt --
+ "Uebersetzt: devices 4" waere an dieser Stelle einfach falsch.
+ """
+ zeilen = [f"{titel}:"]
zeilen += [f" {key}: {value}" for key, value in sorted(self.counts.items())] or [
" nichts"
]
@@ -415,3 +421,266 @@ def convert(legacy: dict[str, Any]) -> tuple[dict[str, Any], Report]:
"Beim Export im alten TESM diese Kategorien mitwaehlen."
)
return payload, report
+
+
+def _eine(conn: Any, sql: str, args: tuple[Any, ...] = ()) -> Any:
+ """Eine Zeile oder ``None`` -- die alte Datenbank kennt kein Database."""
+ return conn.execute(sql, args).fetchone()
+
+
+# --------------------------------------------------------------------------- #
+# Alte Installation lesen
+# --------------------------------------------------------------------------- #
+
+#: Dateinamen der Vorgaengerfassung, relativ zu ihrem Verzeichnis.
+LEGACY_DB_NAME = "sqlite.db"
+LEGACY_FERNET_NAME = "fernet.key"
+
+#: Genau die Abfragen des alten Exports. Sie stehen hier als Text und nicht als
+#: Nachbau: was der Vorgaenger exportiert hat, ist das, was uebersetzbar ist.
+_ABFRAGEN: dict[str, str] = {
+ "devices": """
+ SELECT devices.mac, devices.ip, devices.port, devices.name,
+ devices.switch_hostname, devices.is_active, devices.ssh_port,
+ credentials.name AS credential_name
+ FROM devices LEFT JOIN credentials ON credentials.id = devices.credential_id
+ WHERE devices.deleted_at IS NULL
+ """,
+ "switches": """
+ SELECT switches.hostname, switches.ip, switches.ssh_port,
+ credentials.name AS credential_name
+ FROM switches LEFT JOIN credentials ON credentials.id = switches.credential_id
+ WHERE switches.deleted_at IS NULL
+ """,
+}
+
+
+def _oeffne_alt(pfad: Path) -> Any:
+ """Oeffnet die alte Datenbank **nur lesend**.
+
+ Der alte Dienst laeuft womoeglich noch. Ein Schreibzugriff waere hier in
+ jedem Fall falsch, und ``mode=ro`` macht daraus einen Fehler statt einer
+ Moeglichkeit.
+ """
+ import sqlite3
+
+ if not pfad.is_file():
+ raise LegacyTransferError(f"Keine Datenbank gefunden: {pfad}")
+ conn = sqlite3.connect(f"file:{pfad.as_posix()}?mode=ro", uri=True)
+ conn.row_factory = sqlite3.Row
+ return conn
+
+
+def _tabellen(conn: Any) -> set[str]:
+ return {
+ str(row[0])
+ for row in conn.execute("SELECT name FROM sqlite_master WHERE type='table'").fetchall()
+ }
+
+
+def _entschluessler(root: Path, report: Report) -> Any:
+ """Gibt eine Funktion zurueck, die ein altes Passwort entschluesselt.
+
+ Fehlt der Schluessel, bleibt der Datensatz erhalten und das Passwort leer.
+ Ein Zugangsdatensatz ohne Passwort ist nachtragbar; ein abgebrochenes
+ Upgrade ist es nicht.
+ """
+ schluesseldatei = root / LEGACY_FERNET_NAME
+ if not schluesseldatei.is_file():
+ report.note(
+ f"{LEGACY_FERNET_NAME} nicht gefunden -- die Passwoerter der Zugangsdaten "
+ "bleiben leer und sind nachzutragen."
+ )
+ return lambda _token: ""
+
+ from cryptography.fernet import Fernet, InvalidToken
+
+ try:
+ fernet = Fernet(schluesseldatei.read_bytes().strip())
+ except (ValueError, TypeError):
+ report.note(
+ f"{LEGACY_FERNET_NAME} ist unlesbar -- die Passwoerter der Zugangsdaten "
+ "bleiben leer und sind nachzutragen."
+ )
+ return lambda _token: ""
+
+ gescheitert: list[int] = []
+
+ def entschluesseln(token: Any) -> str:
+ if not token:
+ return ""
+ try:
+ return fernet.decrypt(str(token).encode("utf-8")).decode("utf-8")
+ except (InvalidToken, ValueError):
+ gescheitert.append(1)
+ if len(gescheitert) == 1:
+ report.note(
+ "Mindestens ein Passwort liess sich mit dem gefundenen Schluessel "
+ "nicht entschluesseln -- es bleibt leer."
+ )
+ return ""
+
+ return entschluesseln
+
+
+def _alt_dhcp(conn: Any, vorhanden: set[str]) -> dict[str, Any] | None:
+ """Baut den DHCP-Abschnitt so, wie der alte Export ihn geschrieben hat."""
+ if "dhcp_subnets" not in vorhanden:
+ return None
+
+ einstellungen: dict[str, Any] = {}
+ if "settings" in vorhanden:
+ # Nur die Schluessel holen, die der Uebersetzer kennt -- der alte
+ # Einstellungsbestand ist gross und heisst groesstenteils anders.
+ for key in DHCP_SETTING_KEYS:
+ row = _eine(conn, "SELECT value FROM settings WHERE key=?", (key,))
+ if row and str(row["value"] or ""):
+ einstellungen[key] = row["value"]
+
+ subnets = [
+ dict(row)
+ for row in conn.execute(
+ "SELECT interface, range_start, range_end, gateway, dns, sort_order, enabled "
+ "FROM dhcp_subnets"
+ ).fetchall()
+ ]
+ reservations = (
+ [
+ dict(row)
+ for row in conn.execute(
+ "SELECT mac, ip, name FROM dhcp_manual_reservations"
+ ).fetchall()
+ ]
+ if "dhcp_manual_reservations" in vorhanden
+ else []
+ )
+
+ options: list[dict[str, Any]] = []
+ if "dhcp_option_defs" in vorhanden:
+ for definition in conn.execute(
+ "SELECT id, code, name, type, description, is_standard FROM dhcp_option_defs"
+ ).fetchall():
+ werte = (
+ conn.execute(
+ "SELECT device_mac, value FROM dhcp_option_values WHERE option_def_id=?",
+ (definition["id"],),
+ ).fetchall()
+ if "dhcp_option_values" in vorhanden
+ else []
+ )
+ if not werte and definition["is_standard"]:
+ continue
+ options.append(
+ {
+ "code": definition["code"],
+ "name": definition["name"],
+ "type": definition["type"],
+ "description": definition["description"],
+ "is_standard": definition["is_standard"],
+ "values": [
+ {"device_mac": w["device_mac"], "value": w["value"]} for w in werte
+ ],
+ }
+ )
+
+ if not (einstellungen or subnets or reservations or options):
+ return None
+ return {
+ "settings": einstellungen,
+ "subnets": subnets,
+ "reservations": reservations,
+ "options": options,
+ }
+
+
+def read_installation(root: Path) -> tuple[dict[str, Any], Report]:
+ """Liest eine alte Installation und gibt deren Nutzlast zurueck.
+
+ Ergebnis ist dieselbe Struktur, die ``open_legacy`` aus einer Exportdatei
+ liefert -- der Uebersetzer dahinter bleibt also derselbe. Das ist der
+ Unterschied zum Weg ueber die Oberflaeche: fuer ein Upgrade darf niemand
+ sich anmelden, eine Kategorie anhaken und eine Datei herunterladen muessen.
+
+ Gelesen wird ausschliesslich; die alte Installation bleibt unberuehrt.
+ """
+ root = Path(root)
+ report = Report()
+ conn = _oeffne_alt(root / LEGACY_DB_NAME)
+ try:
+ vorhanden = _tabellen(conn)
+ payload: dict[str, Any] = {}
+
+ for abschnitt, sql in _ABFRAGEN.items():
+ tabelle = abschnitt
+ if tabelle not in vorhanden:
+ report.note(f"Tabelle {tabelle!r} fehlt in der alten Datenbank.")
+ continue
+ payload[abschnitt] = [dict(row) for row in conn.execute(sql).fetchall()]
+
+ if "credentials" in vorhanden:
+ entschluesseln = _entschluessler(root, report)
+ payload["credentials"] = [
+ {
+ "name": row["name"],
+ "username": row["username"],
+ "password": entschluesseln(row["password"]),
+ "category": row["category"],
+ }
+ for row in conn.execute(
+ "SELECT name, username, password, category FROM credentials "
+ "WHERE deleted_at IS NULL"
+ ).fetchall()
+ ]
+ else:
+ report.note("Tabelle 'credentials' fehlt in der alten Datenbank.")
+
+ dhcp = _alt_dhcp(conn, vorhanden)
+ if dhcp:
+ payload["dhcp"] = dhcp
+ finally:
+ conn.close()
+
+ payload = {key: value for key, value in payload.items() if value}
+ if not payload:
+ raise LegacyTransferError(
+ f"In {root} stehen keine uebernehmbaren Daten (Clients, Switche, "
+ "Zugangsdaten, DHCP)."
+ )
+ for abschnitt, inhalt in payload.items():
+ report.took(abschnitt, len(inhalt) if isinstance(inhalt, list) else 1)
+ return payload, report
+
+
+def seal_legacy(payload: dict[str, Any], passphrase: str) -> str:
+ """Schreibt eine Nutzlast im **alten** Umschlag.
+
+ Damit ist die Datei genau das, was der Vorgaenger selbst exportiert haette
+ -- lesbar fuer ``open_legacy`` und fuer den Import des alten TESM. Das ist
+ beim Upgrade der Unterschied zwischen einem Zwischenschritt und einem
+ Rueckweg: geht danach etwas schief, laesst sich diese Datei in die alte
+ Installation zurueckspielen.
+ """
+ import secrets
+ from datetime import datetime
+
+ from cryptography.fernet import Fernet
+ from cryptography.hazmat.primitives import hashes
+ from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
+
+ if len(passphrase) < 8:
+ raise LegacyTransferError("Die Passphrase braucht mindestens 8 Zeichen.")
+
+ salt = secrets.token_bytes(16)
+ kdf = PBKDF2HMAC(
+ algorithm=hashes.SHA256(), length=32, salt=salt, iterations=LEGACY_KDF_ITERATIONS
+ )
+ key = base64.urlsafe_b64encode(kdf.derive(passphrase.encode("utf-8")))
+ token = Fernet(key).encrypt(json.dumps(payload).encode("utf-8"))
+ umschlag = {
+ "app": LEGACY_APP_TAG,
+ "version": 2,
+ "exported_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
+ "salt": base64.urlsafe_b64encode(salt).decode("utf-8"),
+ "payload": token.decode("utf-8"),
+ }
+ return json.dumps(umschlag, indent=2, ensure_ascii=False) + "\n"
diff --git a/apps/tesm/src/tesm/templates/tesm/device_detail.html b/apps/tesm/src/tesm/templates/tesm/device_detail.html
index bc6d060..bf41cbf 100644
--- a/apps/tesm/src/tesm/templates/tesm/device_detail.html
+++ b/apps/tesm/src/tesm/templates/tesm/device_detail.html
@@ -115,7 +115,7 @@
{% else %}{{ pill('fehlgeschlagen', 'danger') }}{% endif %}
{{ event.duration_ms | millis }} |
- {{ event.detail }} |
+ {{ event.detail | audit_detail }} |
{% endfor %}
diff --git a/deploy/upgrade.sh b/deploy/upgrade.sh
new file mode 100644
index 0000000..2feca0b
--- /dev/null
+++ b/deploy/upgrade.sh
@@ -0,0 +1,288 @@
+#!/usr/bin/env bash
+# =============================================================================
+# Upgrade vom Stand SONNET5 (Version 1.x) auf diese Fassung -- mit Datenuebernahme.
+#
+# Aufruf (aus dem entpackten Release-Verzeichnis):
+# sudo bash deploy/upgrade.sh --app tesm
+# sudo bash deploy/upgrade.sh --app tesm-license
+# sudo bash deploy/upgrade.sh --app tesm --instance opus --source /srv/tesm
+#
+# Der Ablauf ist die Vorgabe und braucht keine Schalter:
+#
+# 1. **Sichern.** Vollstaendige Kopie der alten Installation nach
+# /srv/-upgrade-/alt. Erst danach wird irgendetwas
+# angefasst -- und aus dieser Kopie stammen auch alle weiteren Schritte.
+# 2. **Altes System abschalten.** Dienst und Timer werden gestoppt und
+# deaktiviert, damit sie nach einem Neustart nicht zurueckkommen.
+# 3. **Neu installieren.** install.sh dieses Verzeichnisses.
+# 4. **Exportieren.** Aus der Kopie entsteht eine Exportdatei im **alten**
+# Format -- genau die, die der Vorgaenger selbst geschrieben haette.
+# 5. **Uebersetzen.** Daraus eine Datei im neuen Format.
+# 6. **Einspielen.** Diese Datei geht durch denselben Import wie eine
+# Sicherung.
+#
+# Warum die Sicherung vor dem Export steht und nicht danach: laeuft das
+# Upgrade **in dasselbe Verzeichnis** (also ohne --instance), ueberschreibt
+# install.sh die alte Installation. Wer erst exportiert und dann sichert,
+# sichert im schlechtesten Fall nichts mehr. Gelesen wird ausschliesslich aus
+# der Kopie -- das Ergebnis ist dasselbe, die Reihenfolge ist die sichere.
+#
+# Der Lizenzserver kennt Schritt 4 und 5 nicht: seine Uebernahme liest die alte
+# Datenbank unmittelbar und **signiert jede Lizenz neu**. Dabei wird der alte
+# Signaturschluessel uebernommen -- sonst muesste jeder Kunde einen
+# Ausstellerwechsel bestaetigen. Wer das nicht will: --new-key. Dort ist die
+# Kopie aus Schritt 1 der Export.
+# =============================================================================
+set -euo pipefail
+
+APP=""
+INSTANCE=""
+SOURCE=""
+PASSPHRASE=""
+ASSUME_YES=0
+DRY_RUN=0
+WITH_DATA=1
+STOP_OLD=1
+ADOPT_KEY=1
+EXTRA=()
+
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+REPO_DIR="$(cd "$SCRIPT_DIR/.." && pwd)"
+
+COLOR_OK=$'\033[32m'; COLOR_WARN=$'\033[33m'; COLOR_ERR=$'\033[31m'; COLOR_OFF=$'\033[0m'
+step() { printf '\n%s==>%s %s\n' "$COLOR_OK" "$COLOR_OFF" "$*"; }
+warn() { printf '%s !!%s %s\n' "$COLOR_WARN" "$COLOR_OFF" "$*"; }
+fail() { printf '%s !!%s %s\n' "$COLOR_ERR" "$COLOR_OFF" "$*" >&2; exit 1; }
+info() { printf ' %s\n' "$*"; }
+
+usage() {
+ cat <<'USAGE'
+Verwendung: upgrade.sh --app [Optionen]
+
+ --app Welche Anwendung (Pflicht)
+ --source Verzeichnis der alten Installation
+ (Vorgabe: /srv/tesm bzw. /srv/tesm-license)
+ --instance Neue Installation als benannte Instanz anlegen, statt
+ das alte Verzeichnis zu ueberschreiben. Damit bleibt die
+ alte Installation liegen und ist zurueckholbar.
+ --passphrase Passphrase der Zwischendateien. Ohne Angabe wird eine
+ erzeugt und in der Sicherung abgelegt.
+ --no-data Nur installieren, keine Datenuebernahme
+ --keep-old-running Alten Dienst nicht stoppen (nur mit --instance sinnvoll)
+ --new-key Nur Lizenzserver: einen neuen Signaturschluessel
+ erzeugen, statt den alten zu uebernehmen. Dann muessen
+ alle Kunden den Ausstellerwechsel bestaetigen -- die
+ Vorgabe ist deshalb die Uebernahme.
+ --dry-run Nur zeigen, was geschehen wuerde
+ --yes Keine Rueckfrage
+ -h, --help Diese Hilfe
+
+Alles weitere wird unveraendert an install.sh weitergegeben, etwa --https
+oder --domain.
+
+Beispiele:
+ # In dasselbe Verzeichnis, alte Installation wird ersetzt:
+ sudo bash deploy/upgrade.sh --app tesm
+
+ # Daneben, zum Vergleich, alte Installation bleibt unangetastet:
+ sudo bash deploy/upgrade.sh --app tesm --instance opus --keep-old-running
+USAGE
+}
+
+while [[ $# -gt 0 ]]; do
+ case "$1" in
+ --app) APP="${2:?}"; shift 2 ;;
+ --source) SOURCE="${2:?}"; shift 2 ;;
+ --instance) INSTANCE="${2:?}"; shift 2 ;;
+ --passphrase) PASSPHRASE="${2:?}"; shift 2 ;;
+ --no-data) WITH_DATA=0; shift ;;
+ --keep-old-running) STOP_OLD=0; shift ;;
+ --new-key) ADOPT_KEY=0; shift ;;
+ --adopt-key) ADOPT_KEY=1; shift ;; # Vorgabe, hier nur zur Deutlichkeit
+ --dry-run) DRY_RUN=1; shift ;;
+ --yes|-y) ASSUME_YES=1; shift ;;
+ -h|--help) usage; exit 0 ;;
+ *) EXTRA+=("$1"); shift ;;
+ esac
+done
+
+[[ "$APP" == "tesm" || "$APP" == "tesm-license" ]] || { usage; fail "--app fehlt oder ist unbekannt"; }
+[[ "$(id -u)" -eq 0 ]] || fail "bitte mit sudo ausfuehren"
+[[ -f "$SCRIPT_DIR/install.sh" ]] || fail "install.sh fehlt neben diesem Skript"
+
+[[ -n "$SOURCE" ]] || SOURCE="/srv/$APP"
+[[ -d "$SOURCE" ]] || fail "Die alte Installation $SOURCE gibt es nicht (--source setzen)"
+[[ -f "$SOURCE/sqlite.db" ]] || fail "In $SOURCE liegt keine sqlite.db -- das ist keine Installation der Vorgaengerfassung"
+
+NAME="$APP"
+[[ -n "$INSTANCE" ]] && NAME="$APP-$INSTANCE"
+ADMIN="$NAME-admin"
+NEW_ROOT="/srv/$NAME"
+IN_PLACE=0
+[[ "$(cd "$SOURCE" && pwd)" == "$NEW_ROOT" ]] && IN_PLACE=1
+
+STAMP="$(date +%Y%m%d-%H%M%S)"
+BACKUP="/srv/$NAME-upgrade-$STAMP"
+ALT_KOPIE="$BACKUP/alt"
+EXPORT_ALT="$BACKUP/export-alt.json"
+IMPORT_NEU="$BACKUP/import-neu.json"
+
+# Alte Einheiten der Vorgaengerfassung. Der Timer gehoert dazu: nur den Dienst
+# zu stoppen liesse ihn beim naechsten Ablauf wieder anlaufen.
+if [[ "$APP" == "tesm" ]]; then
+ ALTE_UNITS=(tesm.service tesm-check.timer tesm-check.service
+ tesm-check-restart.timer tesm-check-restart.service)
+else
+ ALTE_UNITS=(tesm-license.service)
+fi
+
+step "Vorhaben"
+info "Anwendung : $APP${INSTANCE:+ (Instanz $INSTANCE)}"
+info "Alt : $SOURCE"
+info "Neu : $NEW_ROOT$([[ $IN_PLACE -eq 1 ]] && echo ' << dasselbe Verzeichnis')"
+info "Sicherung : $ALT_KOPIE"
+info "Datenuebernahme: $([[ $WITH_DATA -eq 1 ]] && echo ja || echo nein)"
+info "Altes System : $([[ $STOP_OLD -eq 1 ]] && echo 'stoppen und deaktivieren' || echo 'laeuft weiter')"
+if [[ $IN_PLACE -eq 1 && $STOP_OLD -eq 0 ]]; then
+ warn "--keep-old-running bei gleichem Verzeichnis: der alte Dienst laeuft auf Dateien,"
+ warn "die gleich ersetzt werden. Das geht schief. Bitte --instance verwenden."
+fi
+
+if [[ $DRY_RUN -eq 1 ]]; then
+ step "--dry-run: hier ist Schluss."
+ exit 0
+fi
+
+if [[ $ASSUME_YES -eq 0 ]]; then
+ read -r -p "Fortfahren? [j/N] " answer
+ [[ "$answer" =~ ^[jJyY]$ ]] || { echo "Abgebrochen."; exit 0; }
+fi
+
+# -- 1. Sichern -------------------------------------------------------------- #
+step "Sichern"
+mkdir -p "$ALT_KOPIE"
+chmod 700 "$BACKUP"
+# -a: Rechte und Zeiten bleiben, damit die Kopie auch als Rueckweg taugt.
+# Das venv der alten Installation bleibt aussen vor -- es ist nachbaubar,
+# gross, und seine Shebangs zeigen ohnehin auf den alten Pfad.
+tar -C "$SOURCE" --exclude=venv --exclude=__pycache__ -cf - . | tar -C "$ALT_KOPIE" -xf -
+info "$(du -sh "$ALT_KOPIE" | cut -f1) kopiert (ohne venv)"
+[[ -f "$ALT_KOPIE/sqlite.db" ]] || fail "In der Kopie fehlt die Datenbank -- Abbruch"
+
+# -- 2. Altes System abschalten ---------------------------------------------- #
+if [[ $STOP_OLD -eq 1 ]]; then
+ step "Altes System abschalten"
+ for unit in "${ALTE_UNITS[@]}"; do
+ if systemctl list-unit-files "$unit" --no-legend 2>/dev/null | grep -q .; then
+ systemctl disable --now "$unit" >/dev/null 2>&1 || true
+ info "$unit gestoppt und deaktiviert"
+ fi
+ done
+else
+ step "Altes System bleibt in Betrieb"
+fi
+
+# -- 3. Neu installieren ----------------------------------------------------- #
+step "Neu installieren"
+INSTALL_ARGS=(--app "$APP" --yes)
+[[ -n "$INSTANCE" ]] && INSTALL_ARGS+=(--instance "$INSTANCE")
+bash "$SCRIPT_DIR/install.sh" "${INSTALL_ARGS[@]}" ${EXTRA[0]+"${EXTRA[@]}"}
+
+if [[ $WITH_DATA -eq 0 ]]; then
+ step "Fertig -- ohne Datenuebernahme"
+ info "Sicherung der alten Installation: $ALT_KOPIE"
+ exit 0
+fi
+
+command -v "$ADMIN" >/dev/null 2>&1 || fail "$ADMIN nicht gefunden -- die Installation ist unvollstaendig"
+
+if [[ -z "$PASSPHRASE" ]]; then
+ PASSPHRASE="$(head -c 24 /dev/urandom | base64 | tr -d '=+/' | cut -c1-24)"
+ # Die Passphrase liegt neben den Dateien, die sie schuetzt. Das ist hier kein
+ # Widerspruch: das alte Format verlangt eine, das Verzeichnis gehoert root
+ # allein (0700), und ohne sie waeren die beiden Dateien spaeter unbrauchbar --
+ # auch als Rueckweg.
+ printf '%s\n' "$PASSPHRASE" > "$BACKUP/passphrase.txt"
+ chmod 600 "$BACKUP/passphrase.txt"
+fi
+
+if [[ "$APP" == "tesm" ]]; then
+ # -- 4. Exportieren -------------------------------------------------------- #
+ step "Daten der alten Installation lesen"
+ "$ADMIN" legacy-export --source "$ALT_KOPIE" --out "$EXPORT_ALT" \
+ --passphrase "$PASSPHRASE" --force
+ chmod 600 "$EXPORT_ALT"
+
+ # -- 5. Uebersetzen -------------------------------------------------------- #
+ step "Uebersetzen"
+ "$ADMIN" convert-legacy "$EXPORT_ALT" --ausgabe "$IMPORT_NEU" \
+ --passphrase "$PASSPHRASE" --neue-passphrase "$PASSPHRASE" --force
+ chmod 600 "$IMPORT_NEU"
+
+ # -- 6. Einspielen --------------------------------------------------------- #
+ step "Einspielen"
+ "$ADMIN" legacy-import "$IMPORT_NEU" --passphrase "$PASSPHRASE" --actor upgrade
+else
+ # -- 4.-6. in einem: die Uebernahme liest die alte Datenbank unmittelbar --- #
+ step "Lizenzbestand uebernehmen"
+ # Der alte Signaturschluessel wird uebernommen. Fuer ein Upgrade ist das
+ # die einzige sinnvolle Vorgabe: es ist derselbe Server mit neuerer Software,
+ # und ohne den alten Schluessel muesste jeder Kunde den Ausstellerwechsel
+ # bestaetigen. Ueberschrieben wird dabei nichts -- die Uebernahme lehnt ab,
+ # wenn die neue Installation schon einen eigenen Schluessel hat.
+ MIGRATE_ARGS=(migrate-legacy --source "$ALT_KOPIE")
+ [[ $ADOPT_KEY -eq 1 ]] && MIGRATE_ARGS+=(--adopt-key)
+ "$ADMIN" "${MIGRATE_ARGS[@]}"
+ if [[ $ADOPT_KEY -eq 0 ]]; then
+ warn "--new-key: es gilt ein neuer Signaturschluessel. Die Kunden muessen den"
+ warn "Ausstellerwechsel einmal bestaetigen."
+ fi
+fi
+
+# -- Abschluss --------------------------------------------------------------- #
+step "Fertig"
+info "Sicherung der alten Installation : $ALT_KOPIE"
+if [[ "$APP" == "tesm" ]]; then
+ info "Export im alten Format : $EXPORT_ALT"
+ info "Uebersetzte Datei : $IMPORT_NEU"
+ [[ -f "$BACKUP/passphrase.txt" ]] && info "Passphrase beider Dateien : $BACKUP/passphrase.txt"
+fi
+cat <-.tar.gz`.
+### Upgrade vom Stand 1.x (Branch `SONNET5`)
+
+Ein eigenes Skript, weil dabei mehr passiert als ein Dateiabgleich: die alte
+Fassung hat ein anderes Schema, andere Rechteschlüssel und ein anderes
+Exportformat.
+
+```bash
+sudo bash deploy/upgrade.sh --app tesm
+sudo bash deploy/upgrade.sh --app tesm-license
+```
+
+Der Ablauf ist die Vorgabe und braucht keine Schalter:
+
+| | |
+|---|---|
+| 1. Sichern | Vollständige Kopie der alten Installation nach `/srv/-upgrade-/alt` (ohne `venv`, Verzeichnis 0700) |
+| 2. Abschalten | Alter Dienst **und seine Timer** werden gestoppt und deaktiviert |
+| 3. Installieren | `install.sh` dieses Verzeichnisses |
+| 4. Exportieren | Aus der Kopie entsteht `export-alt.json` im **alten** Format |
+| 5. Übersetzen | Daraus `import-neu.json` im neuen Format |
+| 6. Einspielen | Diese Datei geht durch denselben Import wie eine Sicherung |
+
+Warum die Sicherung **vor** dem Export steht: ohne `--instance` installiert
+`install.sh` in dasselbe Verzeichnis und ersetzt die alte Installation. Wer
+erst exportiert und dann sichert, sichert im schlechtesten Fall nichts mehr.
+Gelesen wird ausschliesslich aus der Kopie -- die alte Installation wird nie
+verändert.
+
+Die beiden Zwischendateien bleiben liegen; ihre Passphrase steht daneben in
+`passphrase.txt` (0600, Verzeichnis 0700). `export-alt.json` ist ein echter
+Export des Vorgängers und taugt deshalb auch als Rückweg.
+
+**Daneben statt darüber.** Mit `--instance` entsteht eine zweite Installation,
+und die alte bleibt vollständig liegen:
+
+```bash
+sudo bash deploy/upgrade.sh --app tesm --instance opus --keep-old-running --port 5100 --http-port 8080
+```
+
+**Was nicht mitkommt** -- und warum:
+
+* **Lokale Konten.** Der alte Hash ist bcrypt, hier gilt Argon2id. Ein Hash
+ ist nicht umrechenbar; das ist sein Zweck. Erstes Konto mit
+ `tesm-admin create-admin`.
+* **Gruppen und Rechte.** Die Rechteschlüssel wurden neu geschnitten.
+* **LDAP, Webserver, Zertifikat.** Von Hand nachziehen.
+
+Beim **Lizenzserver** gibt es Schritt 4 und 5 nicht: die Übernahme liest die
+alte Datenbank unmittelbar, holt Endpunkt und Anbieterdaten aus ihren
+Einstellungen und **signiert jede Lizenz neu**. Der alte Signaturschlüssel wird
+dabei übernommen -- sonst müsste jeder Kunde einen Ausstellerwechsel
+bestätigen; `--new-key` erzwingt das Gegenteil. Der Bericht nennt namentlich,
+wer eine neue Lizenzdatei braucht, denn die Bindung an eine Maschine gilt nur
+für die alte Datei.
+
+Nur die Anwendung, ohne Daten: `--no-data`. Nur zeigen, was geschähe:
+`--dry-run`.
+
---
## 4. HTTPS
diff --git a/packages/tesm-core/pyproject.toml b/packages/tesm-core/pyproject.toml
index 7849ca5..be37739 100644
--- a/packages/tesm-core/pyproject.toml
+++ b/packages/tesm-core/pyproject.toml
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project]
name = "tesm-core"
-version = "2.0.2"
+version = "2.0.3"
description = "Gemeinsames Anwendungsgeruest (Auth, RBAC, Navigation, Design-System, Migrationen) fuer TESM und TESM-Lizenzserver"
requires-python = ">=3.11"
dependencies = [
diff --git a/packages/tesm-core/src/tesm_core/app_factory.py b/packages/tesm-core/src/tesm_core/app_factory.py
index 91fd9ba..3a6d738 100644
--- a/packages/tesm-core/src/tesm_core/app_factory.py
+++ b/packages/tesm-core/src/tesm_core/app_factory.py
@@ -134,7 +134,7 @@ def create_base_app(
#: Seitenhintergrund. Der Lizenzserver setzt sie bewusst nicht: die
#: Funktionsliste von TESM waere dort schlicht falsch.
auth_tagline: str = "",
- app_version: str = "2.0.2",
+ app_version: str = "2.0.3",
built_in_trusted_keys: tuple[str, ...] = (),
enable_license: bool = True,
template_folder: str | None = None,
diff --git a/packages/tesm-core/src/tesm_core/auth/models.py b/packages/tesm-core/src/tesm_core/auth/models.py
index fba7265..3d29990 100644
--- a/packages/tesm-core/src/tesm_core/auth/models.py
+++ b/packages/tesm-core/src/tesm_core/auth/models.py
@@ -10,6 +10,7 @@ from flask import abort, flash, g, redirect, request, session, url_for
from werkzeug.local import LocalProxy
from ..rbac.model import Action, PermissionTree
+from ..security.sessions import is_automatic
AUTH_SOURCES = ("local", "ldap")
@@ -189,6 +190,14 @@ def _deny(message: str) -> Any:
if not current_user.is_authenticated:
if request.method == "GET" and not request.headers.get("X-Requested-With"):
session["_after_login"] = request.full_path if request.query_string else request.path
+ # Eine automatische Aktualisierung fragt einen Ausschnitt ab und
+ # erwartet HTML. Eine Umleitung auf die Anmeldeseite wuerde sie
+ # dankbar entgegennehmen und eine ganze Seite in eine Kachel setzen --
+ # der Tab bliebe stumm stehen, obwohl die Sitzung abgelaufen ist. 401
+ # sagt es deutlich; der Browser laedt daraufhin die Seite neu und der
+ # Mensch sieht die Anmeldemaske.
+ if is_automatic(request):
+ abort(401, description="Sitzung abgelaufen.")
if request.accept_mimetypes.accept_json and not request.accept_mimetypes.accept_html:
abort(401, description="Anmeldung erforderlich.")
flash("Bitte melden Sie sich an.", "info")
diff --git a/packages/tesm-core/src/tesm_core/security/sessions.py b/packages/tesm-core/src/tesm_core/security/sessions.py
index 7c1c111..ae86fd6 100644
--- a/packages/tesm-core/src/tesm_core/security/sessions.py
+++ b/packages/tesm-core/src/tesm_core/security/sessions.py
@@ -34,6 +34,25 @@ from ..db import Database
SESSION_ID_BYTES = 32
+#: Kopfzeile, mit der der Browser eine Anfrage als **automatisch** kennzeichnet:
+#: ein Aktualisierungsdurchlauf einer Kachel, kein Klick eines Menschen.
+#:
+#: Ohne diese Unterscheidung ist die automatische Abmeldung wirkungslos, und das
+#: war sie auch: ein offener Tab fragt seine Bereiche alle 15 Sekunden neu ab,
+#: jede Abfrage verlaengerte das Leerlauffenster, und niemand wurde je
+#: abgemeldet -- nachweisbar an einer Sitzung, deren ``last_seen_at`` eine
+#: Minute alt war, obwohl seit zehn Stunden niemand am Rechner sass.
+#:
+#: Die Entscheidung gehoert zum Client: nur er weiss, ob ein Zeitgeber oder ein
+#: Mensch die Anfrage ausgeloest hat. Ein Angreifer koennte den Kopf setzen --
+#: er wuerde damit seine *eigene* Sitzung schneller ablaufen lassen.
+AUTO_REFRESH_HEADER = "X-Auto-Refresh"
+
+
+def is_automatic(request: Request) -> bool:
+ """Kam die Anfrage von einem Zeitgeber statt von einem Menschen?"""
+ return request.headers.get(AUTO_REFRESH_HEADER, "") == "1"
+
def _now() -> datetime:
return datetime.now(timezone.utc)
@@ -154,8 +173,17 @@ class SqliteSessionInterface(SessionInterface):
}
# Leerlauffenster verlaengern, aber hoechstens einmal pro Minute
# schreiben -- sonst kostet jeder AJAX-Poll einen Schreibvorgang.
+ #
+ # Eine automatische Abfrage verlaengert gar nichts. Sie darf die
+ # Sitzung benutzen, solange diese lebt, aber sie haelt sie nicht am
+ # Leben: sonst bleibt jeder offene Tab bis in alle Ewigkeit
+ # angemeldet, und "Abmeldung nach 30 Minuten Leerlauf" ist eine
+ # Zusage, die die Anwendung nicht einhaelt. Damit bleibt
+ # ``last_seen_at`` ausserdem, was der Name sagt: der letzte
+ # Zeitpunkt, an dem ein Mensch da war.
last_seen = _parse(row.get("last_seen_at"))
- if last_seen is None or (now - last_seen) > timedelta(seconds=60):
+ veraltet = last_seen is None or (now - last_seen) > timedelta(seconds=60)
+ if veraltet and not is_automatic(request):
with self.database.transaction(conn):
conn.execute(
"UPDATE user_sessions SET last_seen_at=?, idle_expires_at=? WHERE id=?",
diff --git a/packages/tesm-core/src/tesm_core/static/tesm_core/js/app.js b/packages/tesm-core/src/tesm_core/static/tesm_core/js/app.js
index ca478bf..97f1fa5 100644
--- a/packages/tesm-core/src/tesm_core/static/tesm_core/js/app.js
+++ b/packages/tesm-core/src/tesm_core/static/tesm_core/js/app.js
@@ -45,12 +45,24 @@ function csrfToken() {
return document.body?.dataset.csrf || "";
}
-/** fetch mit CSRF-Header und JSON-Erwartung. */
+/**
+ * fetch mit CSRF-Header und JSON-Erwartung.
+ *
+ * `auto: true` kennzeichnet eine Anfrage, die ein Zeitgeber ausgeloest hat und
+ * kein Mensch. Der Server verlaengert das Leerlauffenster der Sitzung dann
+ * nicht -- ohne diese Unterscheidung haelt jeder offene Tab die Anmeldung
+ * unbegrenzt am Leben, und die automatische Abmeldung findet nie statt.
+ *
+ * Der HTTP-Status haengt am geworfenen Fehler, damit der Aufrufer 401
+ * ("Sitzung abgelaufen") von einem Netzfehler unterscheiden kann.
+ */
export async function api(url, options = {}) {
- const opts = { credentials: "same-origin", ...options };
+ const { auto = false, ...rest } = options;
+ const opts = { credentials: "same-origin", ...rest };
opts.headers = {
"X-Requested-With": "fetch",
"X-CSRF-Token": csrfToken(),
+ ...(auto ? { "X-Auto-Refresh": "1" } : {}),
...(options.headers || {}),
};
const response = await fetch(url, opts);
@@ -60,7 +72,9 @@ export async function api(url, options = {}) {
const message =
(payload && typeof payload === "object" && (payload.error || payload.message)) ||
`Anfrage fehlgeschlagen (HTTP ${response.status})`;
- throw new Error(message);
+ const error = new Error(message);
+ error.status = response.status;
+ throw error;
}
return payload;
}
@@ -556,7 +570,10 @@ function initAutoRefresh() {
const tick = async () => {
if (stopped || document.hidden) return schedule();
try {
- const html = await api(region.dataset.refreshUrl, { headers: { Accept: "text/html" } });
+ const html = await api(region.dataset.refreshUrl, {
+ auto: true,
+ headers: { Accept: "text/html" },
+ });
// Eine vollstaendige Seite gehoert nie in einen Ausschnitt. Zeigt die
// URL versehentlich auf eine ganze Seite, waere das Ergebnis eine
// Seite in der Seite -- bei jedem Durchlauf eine Ebene tiefer.
@@ -588,7 +605,16 @@ function initAutoRefresh() {
initRelativeTimes(region);
initCountdowns(region);
}
- } catch {
+ } catch (error) {
+ // 401 heisst: die Sitzung ist abgelaufen. Weiterpollen waere sinnlos,
+ // und stehenbleiben waere irrefuehrend -- die Kacheln zeigten dann
+ // beliebig alte Zahlen, als sei alles in Ordnung. Neu laden bringt die
+ // Anmeldemaske mit ihrem Hinweis.
+ if (error && error.status === 401) {
+ stopped = true;
+ window.location.reload();
+ return;
+ }
/* Netzfehler: beim naechsten Intervall erneut versuchen. */
}
schedule();
diff --git a/packages/tesm-core/src/tesm_core/templates/tesm_core/logs_archive.html b/packages/tesm-core/src/tesm_core/templates/tesm_core/logs_archive.html
index 12f280d..7ac18b9 100644
--- a/packages/tesm-core/src/tesm_core/templates/tesm_core/logs_archive.html
+++ b/packages/tesm-core/src/tesm_core/templates/tesm_core/logs_archive.html
@@ -19,7 +19,7 @@
{{ row.actor }} |
{{ row.action }} |
{{ row.target }} |
- {{ row.detail }} |
+ {{ row.detail | audit_detail }} |
{{ row.hash[:12] }} |
{% endfor %}
diff --git a/packages/tesm-core/src/tesm_core/templates/tesm_core/logs_audit.html b/packages/tesm-core/src/tesm_core/templates/tesm_core/logs_audit.html
index ec55e75..364e7ea 100644
--- a/packages/tesm-core/src/tesm_core/templates/tesm_core/logs_audit.html
+++ b/packages/tesm-core/src/tesm_core/templates/tesm_core/logs_audit.html
@@ -64,7 +64,7 @@
{{ row.actor }} |
{{ row.action }} |
{{ row.target }} |
- {{ row.detail }} |
+ {{ row.detail | audit_detail }} |
{# Ampel: Info grün, Hinweis gelb, Warnung rot. Vorher war Info
grau und Hinweis blau -- dabei war nicht zu sehen, was harmlos
diff --git a/packages/tesm-core/src/tesm_core/ui.py b/packages/tesm-core/src/tesm_core/ui.py
index cc0b0a3..bb88e5a 100644
--- a/packages/tesm-core/src/tesm_core/ui.py
+++ b/packages/tesm-core/src/tesm_core/ui.py
@@ -8,12 +8,13 @@ nichts, was die CSP aufweichen muesste.
from __future__ import annotations
import math
+import re
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
from flask import Flask, url_for
-from markupsafe import Markup
+from markupsafe import Markup, escape
ICONS: dict[str, str] = {
"grid": '',
@@ -263,6 +264,89 @@ def asset_url(app: Flask) -> Any:
return build
+
+# --------------------------------------------------------------------------- #
+# Auditeintraege: Dateinamen anklickbar machen
+# --------------------------------------------------------------------------- #
+
+#: Name einer Mitschrift, wie ``terminal.py`` sie bildet: Zeitstempel, Benutzer,
+#: Ziel. Der Ausdruck laesst bewusst keinen Schraegstrich zu -- der Name kommt
+#: aus einem Auditeintrag, und ein Eintrag ist Text, kein Pfad.
+MITSCHRIFT = re.compile(r"\b(\d{8}-\d{6}-[A-Za-z0-9._-]{1,140}\.log)\b")
+
+#: Unterverzeichnisse unter dem Protokollverzeichnis, in denen nachgesehen wird.
+#: Der leere Eintrag steht fuer das Verzeichnis selbst.
+PROTOKOLL_ORTE = ("terminal", "")
+
+
+def _mitschrift_ziel(name: str) -> str | None:
+ """Der Name relativ zum Protokollverzeichnis -- oder ``None``.
+
+ Ein Verweis auf eine Datei, die es nicht mehr gibt, ist schlechter als
+ keiner: er sieht nach einem Angebot aus und endet in 404. Protokolle werden
+ aufgeraeumt, Auditeintraege bleiben -- der Fall ist die Regel, nicht die
+ Ausnahme.
+ """
+ from .extension import core
+
+ try:
+ wurzel = core().config.log_dir
+ except Exception: # noqa: BLE001 - ausserhalb eines App-Kontexts
+ return None
+ for ort in PROTOKOLL_ORTE:
+ relativ = f"{ort}/{name}" if ort else name
+ try:
+ if (wurzel / relativ).is_file():
+ return relativ
+ except OSError: # pragma: no cover - Verzeichnis nicht lesbar
+ continue
+ return None
+
+
+def audit_detail(text: Any) -> Markup:
+ """Auditeintrag als Text -- Dateinamen darin als Verweis.
+
+ Der Eintrag zur Konsole nennt seine Mitschrift beim Namen. Bisher stand
+ dort nur die Zeichenkette, und wer sehen wollte, was passiert ist, musste
+ sie von Hand in die Protokollansicht uebertragen. Es ist derselbe Weg wie
+ zuvor, nur einen Klick weit.
+
+ Verlinkt wird ausschliesslich, wenn die Datei existiert **und** der
+ Betrachter Protokolldateien lesen darf -- sonst bliebe ein Verweis, der in
+ 403 endet. Der Rest des Eintrags wird maskiert; ein Auditeintrag enthaelt
+ Fremdeingaben (Geraetenamen etwa) und darf niemals als HTML gelten.
+ """
+ from .auth.models import current_user
+
+ roh = "" if text is None else str(text)
+ if not roh or not MITSCHRIFT.search(roh):
+ return Markup(escape(roh))
+ try:
+ erlaubt = bool(current_user.is_authenticated and current_user.has("logs_history.view"))
+ except Exception: # noqa: BLE001 - ausserhalb einer Anfrage
+ erlaubt = False
+ if not erlaubt:
+ return Markup(escape(roh))
+
+ def verweis(name: str) -> str:
+ ziel = _mitschrift_ziel(name)
+ if ziel is None:
+ return str(escape(name))
+ try:
+ url = url_for("logs.history_file", name=ziel)
+ except Exception: # noqa: BLE001 - Route fehlt in dieser Anwendung
+ return str(escape(name))
+ return f'{escape(name)}'
+
+ teile: list[str] = []
+ ende = 0
+ for treffer in MITSCHRIFT.finditer(roh):
+ teile.append(str(escape(roh[ende : treffer.start()])))
+ teile.append(verweis(treffer.group(1)))
+ ende = treffer.end()
+ teile.append(str(escape(roh[ende:])))
+ return Markup("".join(teile))
+
def init_app(app: Flask) -> None:
app.jinja_env.globals.update(
icon=icon,
@@ -293,6 +377,7 @@ def init_app(app: Flask) -> None:
from_json=from_json,
issuer_source=issuer_source,
issuer_source_hint=issuer_source_hint,
+ audit_detail=audit_detail,
)
app.jinja_env.trim_blocks = True
app.jinja_env.lstrip_blocks = True
diff --git a/packages/tesm-licensing/pyproject.toml b/packages/tesm-licensing/pyproject.toml
index 2c4107b..cd0db6c 100644
--- a/packages/tesm-licensing/pyproject.toml
+++ b/packages/tesm-licensing/pyproject.toml
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project]
name = "tesm-licensing"
-version = "2.0.2"
+version = "2.0.3"
description = "Ed25519-basiertes Lizenz- und Aktivierungsprotokoll fuer TESM und TESM-Lizenzserver"
requires-python = ">=3.11"
dependencies = ["cryptography>=42"]
|