Upgrade von 1.x, Leerlaufabmeldung, Mitschrift im Audit

upgrade.sh -- ein Weg vom Stand SONNET5 auf diese Fassung, mit Daten. Der
Ablauf braucht keine Schalter: alte Installation sichern, abschalten, neu
installieren, exportieren, uebersetzen, einspielen. Die Sicherung steht
bewusst vor dem Export -- ohne --instance installiert install.sh in dasselbe
Verzeichnis, und wer erst exportiert und dann sichert, sichert im
schlechtesten Fall nichts mehr. Gelesen wird nur aus der Kopie.

Dafuer zwei neue Befehle: "legacy-export" liest eine alte Installation
unmittelbar (der Vorgaenger kann nur ueber die Weboberflaeche exportieren --
fuer ein Upgrade der falsche Weg, und unmoeglich, wenn die alte Anwendung nicht
mehr startet) und schreibt eine Datei im alten Format, die deshalb auch als
Rueckweg taugt. "legacy-import" liest eine Datei im neuen Format ein, ueber
transfer.apply_import -- dieselbe Funktion wie die Seite "Sicherung", kein
zweiter Schreibpfad.

Am Lizenzserver hat der Lauf gegen echte Altdaten zwei Fehler gezeigt, die
jede Uebernahme unmoeglich machten:

* Der alte Signaturschluessel ist URL-sicher kodiert (- und _). Das neue
  Protokoll prueft Base64 streng -- richtig so, aber beim Lesen einer alten
  Datei brach die Uebernahme mit einem Traceback ueber "kein gueltiges Base64"
  ab. Umkodiert wird jetzt an der Grenze zum Altformat; der kanonische Decoder
  bleibt streng.
* Der Anbieterblock wurde aus den Einstellungen der *neuen* Installation
  gebaut, und ohne Endpunkt brach die Uebernahme ab. Bei einem Upgrade ist die
  neue Installation frisch -- dort steht nichts. Endpunkt und Anbieterdaten
  kommen jetzt aus der alten Installation, aber nur dort, wo neu nichts steht.
  Der alte Signaturschluessel wird uebernommen: sonst muesste jeder Kunde
  einen Ausstellerwechsel bestaetigen.

Leerlaufabmeldung: sie hat nie ausgeloest. Ein offener Tab fragt seine
Kacheln alle 15 Sekunden neu ab, und jede Abfrage verlaengerte das
Leerlauffenster -- nachweisbar an einer Sitzung, deren last_seen_at eine
Minute alt war, obwohl seit zehn Stunden niemand am Rechner sass. Der Client
kennzeichnet automatische Anfragen jetzt mit einem Kopf, der Server
verlaengert daraufhin nichts, und last_seen_at bleibt, was der Name sagt.
Faellt die Sitzung waehrend des Pollens, antwortet der Server mit 401 statt
mit einer Umleitung: der Tab laedt neu und zeigt die Anmeldemaske, statt
stumm mit alten Zahlen stehenzubleiben.

Im Audit ist der Name der Mitschrift jetzt ein Verweis auf die Datei -- es ist
derselbe Weg wie zuvor, nur einen Klick weit. Verlinkt wird nur, wenn die
Datei existiert und der Betrachter Protokolle lesen darf; der Rest des
Eintrags wird maskiert, denn ein Auditeintrag enthaelt Fremdeingaben.

Ausserdem: Version 2.0.3.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
2026-09-03 11:43:02 +02:00
co-authored by Claude Opus 5
parent a8bd5763fb
commit 51ae39dd53
15 changed files with 944 additions and 16 deletions
+1 -1
View File
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project] [project]
name = "tesm" name = "tesm"
version = "2.0.2" version = "2.0.3"
description = "TESM -- Verwaltung und Ueberwachung von PoE-Endgeraeten und Aruba-Switchen" description = "TESM -- Verwaltung und Ueberwachung von PoE-Endgeraeten und Aruba-Switchen"
requires-python = ">=3.11" requires-python = ">=3.11"
dependencies = [ dependencies = [
+165
View File
@@ -243,6 +243,131 @@ def cmd_convert_legacy(args: argparse.Namespace) -> int:
return 0 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: def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser( parser = argparse.ArgumentParser(
prog="tesm-admin", description="Verwaltungswerkzeuge fuer TESM" 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.add_argument("--force", action="store_true", help="Zieldatei ueberschreiben")
legacy.set_defaults(func=cmd_convert_legacy) 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) add_web_setup(sub, cmd_web_setup)
return parser return parser
+271 -2
View File
@@ -27,6 +27,7 @@ from __future__ import annotations
import base64 import base64
import json import json
from dataclasses import dataclass, field from dataclasses import dataclass, field
from pathlib import Path
from typing import Any from typing import Any
#: Runden der alten Schluesselableitung. Steht so im Vorgaenger und darf sich #: Runden der alten Schluesselableitung. Steht so im Vorgaenger und darf sich
@@ -84,8 +85,13 @@ class Report:
if text not in self.notes: if text not in self.notes:
self.notes.append(text) self.notes.append(text)
def as_text(self) -> str: def as_text(self, titel: str = "Uebersetzt") -> str:
zeilen = ["Uebersetzt:"] """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 [ zeilen += [f" {key}: {value}" for key, value in sorted(self.counts.items())] or [
" nichts" " nichts"
] ]
@@ -415,3 +421,266 @@ def convert(legacy: dict[str, Any]) -> tuple[dict[str, Any], Report]:
"Beim Export im alten TESM diese Kategorien mitwaehlen." "Beim Export im alten TESM diese Kategorien mitwaehlen."
) )
return payload, report 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"
@@ -115,7 +115,7 @@
{% else %}{{ pill('fehlgeschlagen', 'danger') }}{% endif %} {% else %}{{ pill('fehlgeschlagen', 'danger') }}{% endif %}
</td> </td>
<td class="numeric fs-sm nowrap">{{ event.duration_ms | millis }}</td> <td class="numeric fs-sm nowrap">{{ event.duration_ms | millis }}</td>
<td class="fs-xs muted break-all">{{ event.detail }}</td> <td class="fs-xs muted break-all">{{ event.detail | audit_detail }}</td>
</tr> </tr>
{% endfor %} {% endfor %}
</tbody> </tbody>
+288
View File
@@ -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/<name>-upgrade-<zeitstempel>/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 <tesm|tesm-license> [Optionen]
--app <name> Welche Anwendung (Pflicht)
--source <pfad> Verzeichnis der alten Installation
(Vorgabe: /srv/tesm bzw. /srv/tesm-license)
--instance <name> Neue Installation als benannte Instanz anlegen, statt
das alte Verzeichnis zu ueberschreiben. Damit bleibt die
alte Installation liegen und ist zurueckholbar.
--passphrase <text> 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 <<TEXT
Von Hand nachzuziehen -- das laesst sich nicht uebersetzen:
* Lokale Konten: der alte Hash ist bcrypt, hier gilt Argon2id. Ein Hash
ist nicht umrechenbar; das ist sein Zweck. Erstes Konto anlegen mit
sudo -u $APP $ADMIN create-admin
* Gruppen und Rechte: die Rechteschluessel wurden neu geschnitten.
* Verzeichnisdienst (LDAP), Webserver und Zertifikat.
TEXT
# Der Rueckweg sieht in den beiden Faellen verschieden aus. Ein Hinweis, der
# das alte Verzeichnis wiederherstellen laesst, obwohl es nie angefasst wurde,
# waere schlimmer als keiner: er wuerde die Daten der neuen Installation mit
# einem alten Stand ueberschreiben.
if [[ $IN_PLACE -eq 1 ]]; then
cat <<TEXT
Zurueck auf den alten Stand -- das Verzeichnis wurde ersetzt, die Kopie
bringt es zurueck:
sudo systemctl disable --now $NAME
sudo rm -rf $SOURCE && sudo mkdir -p $SOURCE
sudo tar -C $ALT_KOPIE -cf - . | sudo tar -C $SOURCE -xf -
# venv der alten Installation neu bauen:
sudo python3 -m venv $SOURCE/venv
sudo $SOURCE/venv/bin/python -m pip install -r $SOURCE/requirements.txt
sudo systemctl enable --now ${ALTE_UNITS[0]}
TEXT
else
cat <<TEXT
Zurueck auf den alten Stand -- $SOURCE wurde nicht angefasst, es genuegt,
die neue Instanz abzuschalten:
sudo systemctl disable --now $NAME $NAME-monitor
sudo rm -f /etc/nginx/sites-enabled/$NAME && sudo systemctl reload nginx
sudo systemctl enable --now ${ALTE_UNITS[0]}
TEXT
fi
+58
View File
@@ -141,6 +141,64 @@ etwa auf eine Spiegelung:
Der erwartete Anhang heisst `<repo>-<tag>.tar.gz`. Der erwartete Anhang heisst `<repo>-<tag>.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/<name>-upgrade-<zeitstempel>/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 ## 4. HTTPS
+1 -1
View File
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project] [project]
name = "tesm-core" 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" description = "Gemeinsames Anwendungsgeruest (Auth, RBAC, Navigation, Design-System, Migrationen) fuer TESM und TESM-Lizenzserver"
requires-python = ">=3.11" requires-python = ">=3.11"
dependencies = [ dependencies = [
@@ -134,7 +134,7 @@ def create_base_app(
#: Seitenhintergrund. Der Lizenzserver setzt sie bewusst nicht: die #: Seitenhintergrund. Der Lizenzserver setzt sie bewusst nicht: die
#: Funktionsliste von TESM waere dort schlicht falsch. #: Funktionsliste von TESM waere dort schlicht falsch.
auth_tagline: str = "", auth_tagline: str = "",
app_version: str = "2.0.2", app_version: str = "2.0.3",
built_in_trusted_keys: tuple[str, ...] = (), built_in_trusted_keys: tuple[str, ...] = (),
enable_license: bool = True, enable_license: bool = True,
template_folder: str | None = None, template_folder: str | None = None,
@@ -10,6 +10,7 @@ from flask import abort, flash, g, redirect, request, session, url_for
from werkzeug.local import LocalProxy from werkzeug.local import LocalProxy
from ..rbac.model import Action, PermissionTree from ..rbac.model import Action, PermissionTree
from ..security.sessions import is_automatic
AUTH_SOURCES = ("local", "ldap") AUTH_SOURCES = ("local", "ldap")
@@ -189,6 +190,14 @@ def _deny(message: str) -> Any:
if not current_user.is_authenticated: if not current_user.is_authenticated:
if request.method == "GET" and not request.headers.get("X-Requested-With"): 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 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: if request.accept_mimetypes.accept_json and not request.accept_mimetypes.accept_html:
abort(401, description="Anmeldung erforderlich.") abort(401, description="Anmeldung erforderlich.")
flash("Bitte melden Sie sich an.", "info") flash("Bitte melden Sie sich an.", "info")
@@ -34,6 +34,25 @@ from ..db import Database
SESSION_ID_BYTES = 32 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: def _now() -> datetime:
return datetime.now(timezone.utc) return datetime.now(timezone.utc)
@@ -154,8 +173,17 @@ class SqliteSessionInterface(SessionInterface):
} }
# Leerlauffenster verlaengern, aber hoechstens einmal pro Minute # Leerlauffenster verlaengern, aber hoechstens einmal pro Minute
# schreiben -- sonst kostet jeder AJAX-Poll einen Schreibvorgang. # 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")) 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): with self.database.transaction(conn):
conn.execute( conn.execute(
"UPDATE user_sessions SET last_seen_at=?, idle_expires_at=? WHERE id=?", "UPDATE user_sessions SET last_seen_at=?, idle_expires_at=? WHERE id=?",
@@ -45,12 +45,24 @@ function csrfToken() {
return document.body?.dataset.csrf || ""; 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 = {}) { 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 = { opts.headers = {
"X-Requested-With": "fetch", "X-Requested-With": "fetch",
"X-CSRF-Token": csrfToken(), "X-CSRF-Token": csrfToken(),
...(auto ? { "X-Auto-Refresh": "1" } : {}),
...(options.headers || {}), ...(options.headers || {}),
}; };
const response = await fetch(url, opts); const response = await fetch(url, opts);
@@ -60,7 +72,9 @@ export async function api(url, options = {}) {
const message = const message =
(payload && typeof payload === "object" && (payload.error || payload.message)) || (payload && typeof payload === "object" && (payload.error || payload.message)) ||
`Anfrage fehlgeschlagen (HTTP ${response.status})`; `Anfrage fehlgeschlagen (HTTP ${response.status})`;
throw new Error(message); const error = new Error(message);
error.status = response.status;
throw error;
} }
return payload; return payload;
} }
@@ -556,7 +570,10 @@ function initAutoRefresh() {
const tick = async () => { const tick = async () => {
if (stopped || document.hidden) return schedule(); if (stopped || document.hidden) return schedule();
try { 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 // Eine vollstaendige Seite gehoert nie in einen Ausschnitt. Zeigt die
// URL versehentlich auf eine ganze Seite, waere das Ergebnis eine // URL versehentlich auf eine ganze Seite, waere das Ergebnis eine
// Seite in der Seite -- bei jedem Durchlauf eine Ebene tiefer. // Seite in der Seite -- bei jedem Durchlauf eine Ebene tiefer.
@@ -588,7 +605,16 @@ function initAutoRefresh() {
initRelativeTimes(region); initRelativeTimes(region);
initCountdowns(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. */ /* Netzfehler: beim naechsten Intervall erneut versuchen. */
} }
schedule(); schedule();
@@ -19,7 +19,7 @@
<td class="fs-sm">{{ row.actor }}</td> <td class="fs-sm">{{ row.actor }}</td>
<td class="mono fs-xs">{{ row.action }}</td> <td class="mono fs-xs">{{ row.action }}</td>
<td class="fs-sm">{{ row.target }}</td> <td class="fs-sm">{{ row.target }}</td>
<td class="fs-sm muted">{{ row.detail }}</td> <td class="fs-sm muted">{{ row.detail | audit_detail }}</td>
<td class="mono fs-xs faint">{{ row.hash[:12] }}</td> <td class="mono fs-xs faint">{{ row.hash[:12] }}</td>
</tr> </tr>
{% endfor %} {% endfor %}
@@ -64,7 +64,7 @@
<td class="fs-sm fw-600">{{ row.actor }}</td> <td class="fs-sm fw-600">{{ row.actor }}</td>
<td class="mono fs-xs">{{ row.action }}</td> <td class="mono fs-xs">{{ row.action }}</td>
<td class="fs-sm">{{ row.target }}</td> <td class="fs-sm">{{ row.target }}</td>
<td class="fs-sm muted break-all">{{ row.detail }}</td> <td class="fs-sm muted break-all">{{ row.detail | audit_detail }}</td>
<td> <td>
{# Ampel: Info grün, Hinweis gelb, Warnung rot. Vorher war Info {# Ampel: Info grün, Hinweis gelb, Warnung rot. Vorher war Info
grau und Hinweis blau -- dabei war nicht zu sehen, was harmlos grau und Hinweis blau -- dabei war nicht zu sehen, was harmlos
+86 -1
View File
@@ -8,12 +8,13 @@ nichts, was die CSP aufweichen muesste.
from __future__ import annotations from __future__ import annotations
import math import math
import re
from datetime import datetime, timezone from datetime import datetime, timezone
from pathlib import Path from pathlib import Path
from typing import Any from typing import Any
from flask import Flask, url_for from flask import Flask, url_for
from markupsafe import Markup from markupsafe import Markup, escape
ICONS: dict[str, str] = { ICONS: dict[str, str] = {
"grid": '<rect x="3" y="3" width="7" height="7" rx="1.5"/><rect x="14" y="3" width="7" height="7" rx="1.5"/><rect x="3" y="14" width="7" height="7" rx="1.5"/><rect x="14" y="14" width="7" height="7" rx="1.5"/>', "grid": '<rect x="3" y="3" width="7" height="7" rx="1.5"/><rect x="14" y="3" width="7" height="7" rx="1.5"/><rect x="3" y="14" width="7" height="7" rx="1.5"/><rect x="14" y="14" width="7" height="7" rx="1.5"/>',
@@ -263,6 +264,89 @@ def asset_url(app: Flask) -> Any:
return build 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'<a href="{escape(url)}" title="Mitschrift ansehen">{escape(name)}</a>'
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: def init_app(app: Flask) -> None:
app.jinja_env.globals.update( app.jinja_env.globals.update(
icon=icon, icon=icon,
@@ -293,6 +377,7 @@ def init_app(app: Flask) -> None:
from_json=from_json, from_json=from_json,
issuer_source=issuer_source, issuer_source=issuer_source,
issuer_source_hint=issuer_source_hint, issuer_source_hint=issuer_source_hint,
audit_detail=audit_detail,
) )
app.jinja_env.trim_blocks = True app.jinja_env.trim_blocks = True
app.jinja_env.lstrip_blocks = True app.jinja_env.lstrip_blocks = True
+1 -1
View File
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project] [project]
name = "tesm-licensing" name = "tesm-licensing"
version = "2.0.2" version = "2.0.3"
description = "Ed25519-basiertes Lizenz- und Aktivierungsprotokoll fuer TESM und TESM-Lizenzserver" description = "Ed25519-basiertes Lizenz- und Aktivierungsprotokoll fuer TESM und TESM-Lizenzserver"
requires-python = ">=3.11" requires-python = ">=3.11"
dependencies = ["cryptography>=42"] dependencies = ["cryptography>=42"]