Files
tesm-license/apps/tesm-license/src/tesm_license/cli.py
T
alientimandClaude Opus 5 2dea198bcc 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>
2026-09-03 11:43:04 +02:00

494 lines
20 KiB
Python

"""Kommandozeilenwerkzeuge des Lizenzservers.
Enthaelt insbesondere ``bootstrap`` -- die eigene Lizenz der Wurzelinstanz.
Warum das ein lokales Skript ist und keine Funktion in der Oberflaeche: die
Master-Lizenz ist die einzige Lizenz, fuer die es **keine hoehere Instanz gibt**,
die sie bestaetigen koennte. Ein Aktivierungs-Handshake gegen sich selbst waere
Theater. Das Skript signiert die Datei mit dem eigenen Schluessel und setzt den
Aktivierungszustand direkt -- und nur so kann eine Lizenzdatei je die Rolle
``master`` tragen. Die Ticket-Oberflaeche bietet diese Rolle an keiner Stelle an.
"""
from __future__ import annotations
import argparse
import getpass
import json
import sys
from pathlib import Path
from typing import Any
import tesm_licensing as lic
import tesm_licensing_server as licsrv
from tesm_core import audit
from tesm_core.cli_web import add_web_setup, run_web_setup
from tesm_core.db import Database
from tesm_core.extension import core
from tesm_core.rbac import service as rbac
from tesm_core.security import passwords
from . import create_app, master_key_path
from .services import legacy_import, master_key
def _app_and_core() -> tuple[Any, Any]:
app = create_app()
return app, core(app)
def cmd_init_key(args: argparse.Namespace) -> int:
app, extension = _app_and_core()
path = master_key_path(extension.config)
if path.exists() and not args.force:
key = master_key.load(path)
print(f"Signaturschluessel existiert bereits: {key.key_id} ({path})")
print("Ein Austausch macht ALLE bereits ausgestellten Lizenzen unpruefbar.")
return 0
if path.exists() and args.force:
print(
"ABBRUCH: --force auf einen bestehenden Signaturschluessel wird nicht unterstuetzt.\n"
"Ein neuer Schluessel entwertet jede ausgestellte Lizenz. Bitte die Datei bewusst\n"
"von Hand beiseite legen, wenn das wirklich gewollt ist.",
file=sys.stderr,
)
return 2
key = master_key.ensure(path)
print(f"Signaturschluessel erzeugt: {key.key_id}")
print(f"Datei: {path}")
print("Bitte SOFORT sichern -- ohne sie ist keine ausgestellte Lizenz mehr pruefbar.")
print(f"Oeffentlicher Schluessel (zum Pinnen beim Kunden):\n{key.public_key}")
return 0
def cmd_bootstrap(args: argparse.Namespace) -> int:
"""Stellt die eigene Lizenz dieser Wurzelinstanz aus."""
app, extension = _app_and_core()
key = master_key.ensure(master_key_path(extension.config))
with app.app_context(), extension.database.session() as conn:
vendor = {
"name": str(extension.settings.get(conn, "vendor_name") or "Eigenbetrieb"),
"email": str(extension.settings.get(conn, "vendor_email") or ""),
"phone": str(extension.settings.get(conn, "vendor_phone") or ""),
"address": str(extension.settings.get(conn, "vendor_address") or ""),
"logo_base64": "",
}
bundle, _ = licsrv.issue_license(
customer={"name": args.name, "contact_email": vendor["email"]},
license_type="enterprise",
modules=[],
valid_days=int(args.days),
master_private_key_b64=key.private_key,
master_public_key_b64=key.public_key,
# Bewusst leer: es gibt keinen uebergeordneten Endpunkt. Damit
# laeuft der Handshake fuer diese Lizenz nie ins Netz.
master_endpoint="",
vendor=vendor,
role="master",
)
document, client_key = lic.split_bundle(bundle)
config = extension.config
config.license_path.parent.mkdir(parents=True, exist_ok=True)
config.license_path.write_text(
json.dumps(document, indent=2, sort_keys=True, ensure_ascii=False) + "\n",
encoding="utf-8",
)
try:
config.license_path.chmod(0o640)
except OSError:
pass
from tesm_core.keystore import write_private_file
write_private_file(
config.license_key_path,
json.dumps({"license_id": document["license_id"], "client_key": client_key}) + "\n",
)
with extension.database.transaction(conn):
conn.execute(
"INSERT INTO trusted_issuer_keys (key_id, public_key, label, source, added_at, "
"added_by) VALUES (?,?,?,'bootstrap',datetime('now'),'cli') "
"ON CONFLICT(key_id) DO NOTHING",
(key.key_id, key.public_key, "Eigener Signaturschluessel"),
)
# Es gibt keine hoehere Instanz, die diese Lizenz aktivieren koennte.
extension.settings.set(conn, "license_activation_status", "active", actor="cli")
extension.settings.set(conn, "license_pending_request", "", actor="cli")
audit.log(
conn,
actor="cli",
action="own_license.bootstrapped",
target=document["license_id"],
detail=f"Rolle master, {args.days} Tage, Schluessel {key.key_id}",
severity="warning",
)
state = extension.license.reload(conn) if extension.license else None
print(f"Master-Lizenz ausgestellt: {document['license_id']}")
print(f"Gueltig bis: {document['expires_at']}")
print(f"Datei: {config.license_path}")
if state is not None:
print(f"Zustand: gueltig={state.valid}, aktiviert={state.activated}, Rolle={state.role}")
return 0
def cmd_create_admin(args: argparse.Namespace) -> int:
app, extension = _app_and_core()
username = args.username or input("Benutzername: ").strip()
if not username:
print("Kein Benutzername angegeben.", file=sys.stderr)
return 2
password = args.password or getpass.getpass("Passwort: ")
if not args.password and password != getpass.getpass("Passwort wiederholen: "):
print("Die Passwoerter stimmen nicht ueberein.", file=sys.stderr)
return 2
problems = extension.password_policy.check(password, username=username)
if problems and not args.force:
for problem in problems:
print(f"- {problem}", file=sys.stderr)
return 2
with app.app_context(), extension.database.session() as conn:
existing = rbac.find_user_by_username(conn, username, include_deleted=True)
with extension.database.transaction(conn):
if existing is not None:
conn.execute(
"UPDATE users SET password_hash=?, is_admin=1, is_locked=0, deleted_at=NULL, "
"must_change_password=0, updated_at=datetime('now') WHERE id=?",
(passwords.hash_password(password), existing["id"]),
)
action = "aktualisiert"
else:
conn.execute(
"INSERT INTO users (username, password_hash, is_admin, auth_source, "
"created_at, updated_at, password_changed_at) "
"VALUES (?,?,1,'local',datetime('now'),datetime('now'),datetime('now'))",
(username, passwords.hash_password(password)),
)
action = "angelegt"
audit.log(
conn,
actor="cli",
action="user.admin_created",
target=username,
severity="warning",
)
print(f"Administrator {username!r} {action}.")
return 0
def cmd_status(args: argparse.Namespace) -> int:
app, extension = _app_and_core()
with app.app_context(), extension.database.session() as conn:
chain = audit.verify_chain(conn, limit=5000)
counts = {
"kunden": Database.value(conn, "SELECT COUNT(*) FROM license_customers", default=0),
"tickets": Database.value(conn, "SELECT COUNT(*) FROM license_tickets", default=0),
"lizenzen": Database.value(conn, "SELECT COUNT(*) FROM licenses", default=0),
"aktiv": Database.value(
conn, "SELECT COUNT(*) FROM licenses WHERE status='active'", default=0
),
"administratoren": rbac.count_active_admins(conn),
}
state = extension.license.reload(conn) if extension.license else None
key_health = master_key.health(master_key_path(extension.config))
payload = {
"datenbank": str(extension.config.db_path),
"bestand": counts,
"audit_kette": chain,
"signaturschluessel": key_health,
"eigene_lizenz": (
{
"vorhanden": state.present,
"gueltig": state.valid,
"aktiviert": state.activated,
"rolle": state.role,
"fehler": state.error,
}
if state
else None
),
}
if args.json:
print(json.dumps(payload, indent=2, ensure_ascii=False, default=str))
return 0
print(f"Datenbank : {payload['datenbank']}")
for key, value in counts.items():
print(f"{key.capitalize():<17}: {value}")
print(
"Audit-Kette : "
+ ("unversehrt" if chain["ok"] else f"GEBROCHEN bei {chain['broken_at']}")
)
print(
"Signaturschluessel: "
+ (str(key_health.get("key_id")) if key_health.get("present") else "FEHLT")
)
for problem in key_health.get("problems", []):
print(f" ! {problem}", file=sys.stderr)
if state:
print(
"Eigene Lizenz : "
+ ("aktiv" if state.valid and state.activated else "nicht aktiv")
+ f", Rolle {state.role}"
)
if not counts["administratoren"]:
print("\nWARNUNG: kein entsperrtes Administratorkonto vorhanden.", file=sys.stderr)
return 1
return 0
def cmd_housekeeping(_args: argparse.Namespace) -> int:
app, extension = _app_and_core()
from tesm_core.security import ratelimit
from tesm_core.security import sessions as session_store
from .services import activation
with app.app_context(), extension.database.session() as conn:
with extension.database.transaction(conn):
sessions = session_store.prune(conn)
events = ratelimit.prune(conn)
nonces = activation.prune_nonces(conn)
archived = audit.archive_old_rows(conn, extension.audit_archive_dir)
expired_tokens = conn.execute(
"DELETE FROM portal_access_tokens WHERE expires_at < datetime('now', '-7 days')"
).rowcount
print(
f"{sessions} Sitzungen, {events} Ratenereignisse, {nonces} Nonces, "
f"{expired_tokens or 0} Portal-Token entfernt, {len(archived)} Protokolltage archiviert."
)
return 0
def cmd_web_setup(args: argparse.Namespace) -> int:
app, extension = _app_and_core()
return run_web_setup(app, extension, args)
def cmd_migrate_legacy(args: argparse.Namespace) -> int:
"""Uebernimmt Signaturschluessel, Kunden, Tickets und Lizenzen der Vorgaengerfassung."""
root = Path(args.source)
app, extension = _app_and_core()
try:
legacy = legacy_import.open_legacy(root)
except legacy_import.LegacyImportError as exc:
print(str(exc), file=sys.stderr)
return 2
key_path = master_key_path(extension.config)
report = legacy_import.Report()
try:
# -- Schluessel ---------------------------------------------------- #
# Der alte Schluessel liegt als 0600 root. Gelesen wird er nur, wenn er
# wirklich gebraucht wird -- naemlich beim Uebernehmen. Fuer den blossen
# Vergleich genuegt die oeffentliche Haelfte, und fehlt sie, ist das
# kein Grund abzubrechen: der Vergleich ist eine Auskunft, keine
# Voraussetzung.
issuer_changes = False
legacy_public = ""
try:
_, legacy_public = legacy_import.read_legacy_key(root)
except legacy_import.LegacyImportError as exc:
if args.adopt_key:
raise
print(f"Hinweis: alter Signaturschluessel nicht lesbar ({exc}).")
if key_path.is_file():
existing = master_key.load(key_path)
report.key_id = existing.key_id
if legacy_public and existing.public_key == legacy_public:
print(f"Signaturschluessel stimmt bereits ueberein ({existing.key_id}).")
elif not legacy_public:
issuer_changes = True
print(
f"Es wird mit dem vorhandenen Signaturschluessel {existing.key_id} "
"gearbeitet."
)
else:
# Nicht ueberschreiben: jede mit dem vorhandenen Schluessel
# ausgestellte Lizenz waere danach unpruefbar.
issuer_changes = True
print(
f"Diese Installation hat einen eigenen Signaturschluessel "
f"({existing.key_id}); der alte war {lic.key_id(legacy_public)}. "
"Es wird mit dem vorhandenen weitergearbeitet."
)
elif args.adopt_key:
legacy_private, legacy_public = legacy_import.read_legacy_key(root)
report.key_id = legacy_import.adopt_key(legacy_private, legacy_public, key_path)
report.key_adopted = True
print(f"Signaturschluessel uebernommen: {report.key_id}")
else:
print(
"Diese Installation hat noch keinen Signaturschluessel. Mit --adopt-key "
"wird der alte uebernommen -- dann bleibt die Anbieteridentitaet erhalten "
"und Kunden muessen keinen Ausstellerwechsel bestaetigen.",
file=sys.stderr,
)
return 2
key = master_key.load(key_path)
with app.app_context(), extension.database.session() as conn:
# Anbieterdaten und Endpunkt zuerst: bei einem Upgrade ist die neue
# Installation frisch, dort steht also nichts, und die Pruefung
# weiter unten haette jedes Upgrade abgebrochen. Beides gehoert zur
# Anbieteridentitaet -- genau die soll ein Upgrade nicht wechseln.
# Uebernommen wird nur, was hier leer ist.
if not args.dry_run:
with extension.database.transaction(conn):
uebernommen = legacy_import.adopt_vendor_settings(
legacy,
lambda key: extension.settings.get(conn, key),
lambda key, wert: extension.settings.set(
conn, key, wert, actor="migration"
),
)
# Nicht "key" als Laufvariable: darueber steht der geladene
# Signaturschluessel mit demselben Namen, und ihn zu
# ueberschreiben lief in ein AttributeError mitten in der
# Uebernahme.
for einstellung, wert in uebernommen.items():
gekuerzt = wert if len(wert) <= 60 else wert[:57] + "..."
print(f"Uebernommen: {einstellung} = {gekuerzt}")
vendor = {
"name": str(extension.settings.get(conn, "vendor_name") or "Eigenbetrieb"),
"email": str(extension.settings.get(conn, "vendor_email") or ""),
"phone": str(extension.settings.get(conn, "vendor_phone") or ""),
"address": str(extension.settings.get(conn, "vendor_address") or ""),
"logo_base64": "",
}
endpoint = str(extension.settings.get(conn, "master_endpoint") or "")
if not endpoint and not args.dry_run:
print(
"Es ist kein Endpunkt hinterlegt, und die alte Installation nennt "
"auch keinen. Bitte unter Einstellungen -> Anbieter eintragen: ohne "
"ihn koennen uebernommene Lizenzen sich spaeter nicht aktivieren.",
file=sys.stderr,
)
return 2
with extension.database.transaction(conn):
result = legacy_import.import_all(
legacy,
conn,
keystore=extension.keystore,
master_private_key=key.private_key,
master_public_key=key.public_key,
master_endpoint=endpoint,
vendor=vendor,
actor="migration",
reissue=not args.dry_run,
)
result.key_adopted = report.key_adopted
result.key_id = report.key_id
if issuer_changes:
result.warnings.append(
"Die Lizenzen sind mit einem anderen Schluessel signiert als "
"frueher. Kunden, die den alten Aussteller gepinnt haben, muessen "
"beim Einspielen einmal 'Wechsel des Ausstellers ausdruecklich "
"zulassen' bestaetigen."
)
if args.dry_run:
# Nichts festschreiben -- der Lauf war nur zur Ansicht.
raise _DryRun(result)
except _DryRun as preview:
print()
print("== Probelauf, nichts gespeichert ==")
print(preview.report.as_text())
return 0
except legacy_import.LegacyImportError as exc:
print(str(exc), file=sys.stderr)
return 2
finally:
legacy.close()
print()
print(result.as_text())
return 0
class _DryRun(Exception):
"""Bricht die Transaktion eines Probelaufs ab -- ohne Fehler zu sein."""
def __init__(self, report: Any) -> None:
super().__init__("Probelauf")
self.report = report
def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(
prog="tesm-license-admin", description="Verwaltungswerkzeuge des TESM-Lizenzservers"
)
sub = parser.add_subparsers(dest="command", required=True)
init_key = sub.add_parser("init-key", help="Signaturschluessel erzeugen (einmalig)")
init_key.add_argument("--force", action="store_true")
init_key.set_defaults(func=cmd_init_key)
bootstrap = sub.add_parser(
"bootstrap", help="Eigene Master-Lizenz dieser Wurzelinstanz ausstellen"
)
bootstrap.add_argument("--days", type=int, default=3650)
bootstrap.add_argument("--name", default="Lizenzserver (Eigenbetrieb)")
bootstrap.set_defaults(func=cmd_bootstrap)
admin = sub.add_parser("create-admin", help="Administratorkonto anlegen oder zuruecksetzen")
admin.add_argument("--username")
admin.add_argument("--password")
admin.add_argument("--force", action="store_true")
admin.set_defaults(func=cmd_create_admin)
status = sub.add_parser("status", help="Zustand ausgeben")
status.add_argument("--json", action="store_true")
status.set_defaults(func=cmd_status)
house = sub.add_parser("housekeeping", help="Laufzeitdaten aufraeumen")
house.set_defaults(func=cmd_housekeeping)
migrate = sub.add_parser(
"migrate-legacy",
help="Lizenzserver der Vorgaengerfassung uebernehmen",
description=(
"Uebernimmt Signaturschluessel, Kunden, Tickets und Lizenzen aus einer alten "
"Installation. Die alte Installation wird nur gelesen. Jede Lizenz wird im "
"neuen Format neu signiert -- die Kunden brauchen danach eine neue "
"Lizenzdatei, weil die alte Signatur nicht mehr gilt."
),
)
migrate.add_argument(
"--source", default="/srv/tesm-license", help="Verzeichnis der alten Installation"
)
migrate.add_argument(
"--adopt-key",
action="store_true",
help="Den alten Signaturschluessel uebernehmen. Nur moeglich, solange diese "
"Installation noch keinen eigenen hat -- ein vorhandener wird nie ueberschrieben.",
)
migrate.add_argument(
"--dry-run",
action="store_true",
help="Nur anzeigen, was passieren wuerde -- nichts schreiben",
)
migrate.set_defaults(func=cmd_migrate_legacy)
add_web_setup(sub, cmd_web_setup)
return parser
def main(argv: list[str] | None = None) -> int:
args = build_parser().parse_args(argv)
return int(args.func(args))
if __name__ == "__main__": # pragma: no cover
raise SystemExit(main())