"""Uebernahme eines Lizenzservers der Vorgaengerfassung. Die alten Lizenzdateien sind mit der neuen Fassung **nicht** verwendbar, und das ist Absicht: dort wurde ohne Domaenentrennung signiert (die Signatur ueber eine Lizenz war damit auch als Signatur ueber eine Protokollantwort gueltig), es fehlten Format-Version, Sequenznummer und Gueltigkeitsbeginn. Beides gleichzeitig zu akzeptieren hiesse, die Haertung wieder aufzugeben -- eine Uebergangsfrist waere hier eine dauerhaft offene Tuer. Was **sehr wohl** uebernommen wird, ist alles andere: * **Der Signaturschluessel.** Damit bleibt die Anbieteridentitaet dieselbe. Bei Kunden, die den Aussteller bereits gepinnt haben, passt der Schluessel weiterhin -- sie muessen keinen Ausstellerwechsel bestaetigen. * **Kunden, Tickets und Lizenzen** aus der alten Datenbank. Jede uebernommene Lizenz wird im neuen Format **neu signiert**. Der Kunde bekommt dadurch eine neue Lizenzdatei; das laesst sich nicht vermeiden, denn die alte traegt eine Signatur, die es in dieser Form nicht mehr gibt. Der Bericht am Ende sagt genau, wer eine neue Datei braucht. Der Vorgang ist **lesend** gegenueber der alten Installation. Sie wird nicht angefasst, nicht angehalten und nicht veraendert. """ from __future__ import annotations import base64 import binascii import json import sqlite3 from dataclasses import dataclass, field from datetime import datetime, timezone from pathlib import Path from typing import Any import tesm_licensing as lic import tesm_licensing_server as licsrv from tesm_core.db import Database from tesm_core.keystore import Keystore from . import issuance class LegacyImportError(Exception): """Die alte Installation ist nicht lesbar oder passt nicht.""" @dataclass class Report: """Was die Uebernahme getan hat -- vollstaendig und nachpruefbar.""" key_adopted: bool = False key_id: str = "" customers: int = 0 tickets: int = 0 licenses: int = 0 reissued: list[str] = field(default_factory=list) skipped: list[str] = field(default_factory=list) warnings: list[str] = field(default_factory=list) def as_text(self) -> str: lines = [ f"Signaturschluessel uebernommen: {'ja' if self.key_adopted else 'nein'}" + (f" ({self.key_id})" if self.key_id else ""), f"Kunden : {self.customers}", f"Tickets : {self.tickets}", f"Lizenzen : {self.licenses}", ] if self.reissued: lines.append("") lines.append("Neu signiert -- diese Kunden brauchen eine neue Lizenzdatei:") lines.extend(f" * {entry}" for entry in self.reissued) if self.skipped: lines.append("") lines.append("Uebersprungen:") lines.extend(f" * {entry}" for entry in self.skipped) if self.warnings: lines.append("") lines.append("Hinweise:") lines.extend(f" * {entry}" for entry in self.warnings) return "\n".join(lines) def _now() -> str: return datetime.now(timezone.utc).isoformat(timespec="seconds") # --------------------------------------------------------------------------- # # Quelle # --------------------------------------------------------------------------- # def open_legacy(root: Path) -> sqlite3.Connection: """Oeffnet die alte Datenbank **nur lesend**.""" path = Path(root) / "sqlite.db" if not path.is_file(): raise LegacyImportError(f"Unter {path} liegt keine Datenbank der Vorgaengerfassung.") try: conn = sqlite3.connect(f"file:{path}?mode=ro", uri=True) except sqlite3.Error as exc: raise LegacyImportError(f"Die alte Datenbank ist nicht lesbar: {exc}") from exc conn.row_factory = sqlite3.Row tables = {row[0] for row in conn.execute("SELECT name FROM sqlite_master WHERE type='table'")} missing = {"license_customers", "license_tickets", "licenses"} - tables if missing: conn.close() raise LegacyImportError( f"Das sieht nicht nach einem Lizenzserver aus -- es fehlen: {', '.join(sorted(missing))}" ) return conn def _standard_b64(wert: str, feld: str) -> str: """Nimmt Standard- **und** URL-Alphabet an, gibt Standard-Base64 zurueck. Der Vorgaenger kodierte seine Schluessel mit ``urlsafe_b64encode``: dort stehen ``-`` und ``_`` an den Stellen von ``+`` und ``/``. Das neue Protokoll prueft Base64 streng -- richtig so, aber beim Lesen einer alten Datei fuehrte es dazu, dass die Uebernahme mit einem Traceback ueber "kein gueltiges Base64" abbrach, bei jedem echten Altbestand. """ roh = wert.strip() if not roh: raise LegacyImportError(f"Im alten Signaturschluessel fehlt {feld!r}.") # Fehlende Polsterung ergaenzen -- der Vorgaenger schrieb sie mit, andere # Werkzeuge lassen sie weg. gepolstert = roh + "=" * (-len(roh) % 4) # Erst das Standard-, dann das URL-Alphabet -- beide mit Pruefung. Ohne # ``validate`` verschluckt der Decoder unbekannte Zeichen einfach, und aus # "###" wuerde eine leere Ausgabe statt einer Fehlermeldung. versuche = (gepolstert, gepolstert.replace("-", "+").replace("_", "/")) for kandidat in versuche: try: rohbytes = base64.b64decode(kandidat, validate=True) except (ValueError, binascii.Error): continue if len(rohbytes) != 32: raise LegacyImportError( f"{feld} im alten Signaturschluessel ist {len(rohbytes)} Byte lang, " "erwartet werden 32 (Ed25519)." ) return base64.b64encode(rohbytes).decode("ascii") raise LegacyImportError(f"{feld} im alten Signaturschluessel ist kein Base64.") def read_legacy_key(root: Path) -> tuple[str, str]: """Liest den alten Signaturschluessel. Rueckgabe: ``(privat, oeffentlich)``.""" path = Path(root) / "master_signing_key.json" if not path.is_file(): raise LegacyImportError(f"Der alte Signaturschluessel {path} fehlt.") try: data = json.loads(path.read_text(encoding="utf-8")) except (OSError, ValueError) as exc: raise LegacyImportError(f"Der alte Signaturschluessel ist unlesbar: {exc}") from exc # Die Vorgaengerfassung nannte die Felder "private"/"public" und kodierte # sie URL-sicher. private = str(data.get("private") or data.get("private_key") or "") public = str(data.get("public") or data.get("public_key") or "") if not private or not public: raise LegacyImportError("Der alte Signaturschluessel ist unvollstaendig.") private = _standard_b64(private, "private") public = _standard_b64(public, "public") # Alles, was aus der Protokollschicht kommt, wird hier zu einer Meldung im # Klartext. Ein Traceback ueber Base64 mitten in einer Uebernahme sagt dem # Betreiber nichts -- und die Uebernahme bricht ohne --adopt-key gar nicht # ab, sie merkt sich nur den alten Aussteller nicht. try: passt = licsrv.public_from_private(private) == public except (ValueError, TypeError) as exc: raise LegacyImportError(f"Der alte Signaturschluessel ist unbrauchbar: {exc}") from exc if not passt: raise LegacyImportError( "Im alten Signaturschluessel passen privater und oeffentlicher Teil nicht " "zueinander. Die Datei ist beschaedigt." ) return private, public def adopt_key(private: str, public: str, target: Path) -> str: """Uebernimmt den alten Schluessel in das Format der neuen Fassung. Damit bleibt die Anbieteridentitaet erhalten: Kunden, die den Aussteller schon gepinnt haben, sehen weiterhin denselben Schluessel und muessen keinen Wechsel bestaetigen. """ from tesm_core.keystore import write_private_file if Path(target).is_file(): raise LegacyImportError( f"Unter {target} liegt bereits ein Signaturschluessel. Er wird nicht " "ueberschrieben -- das wuerde jede damit ausgestellte Lizenz unbrauchbar " "machen. Fuer eine Uebernahme bitte eine frische Installation verwenden." ) payload = { "private_key": private, "public_key": public, "key_id": lic.key_id(public), "created_at": _now(), "origin": "Uebernommen aus der Vorgaengerfassung.", } write_private_file(Path(target), json.dumps(payload, indent=2, sort_keys=True) + "\n") return lic.key_id(public) # --------------------------------------------------------------------------- # # Uebernahme # --------------------------------------------------------------------------- # def _remaining_days(expires_at: str) -> int: """Restlaufzeit in Tagen, mindestens 1 bei noch gueltiger Lizenz. Die neue Ausstellung rechnet in Tagen ab jetzt. Wer noch drei Monate hatte, soll auch nach der Uebernahme drei Monate haben -- nicht ploetzlich ein volles Jahr und nicht weniger als vorher. """ try: moment = lic.parse_iso(expires_at) except (ValueError, TypeError): return 0 remaining = (moment - datetime.now(timezone.utc)).days return remaining if remaining > 0 else 0 def _modules(raw: Any) -> list[str]: try: values = json.loads(str(raw or "[]")) except ValueError: return [] return sorted(set(str(value) for value in values) & set(lic.ALL_MODULES)) #: Einstellungen, die zur Anbieteridentitaet gehoeren und in beiden Fassungen #: gleich heissen. ``master_endpoint`` ist darunter das entscheidende: ohne ihn #: findet eine uebernommene Lizenz ihren Server spaeter nicht wieder. VENDOR_SETTINGS = ( "master_endpoint", "vendor_name", "vendor_email", "vendor_phone", "vendor_address", "vendor_logo_base64", ) def read_vendor_settings(legacy: sqlite3.Connection) -> dict[str, str]: """Liest die Anbietereinstellungen der alten Installation.""" try: rows = legacy.execute( "SELECT key, value FROM settings WHERE key IN ({})".format( ",".join("?" for _ in VENDOR_SETTINGS) ), VENDOR_SETTINGS, ).fetchall() except sqlite3.Error: # pragma: no cover - alte Fassung ohne settings return {} return {str(row["key"]): str(row["value"] or "") for row in rows if str(row["value"] or "")} def adopt_vendor_settings( legacy: sqlite3.Connection, read: Any, write: Any, ) -> dict[str, str]: """Fuellt leere Anbietereinstellungen aus der alten Installation. ``read(key)`` und ``write(key, wert)`` kommen von aussen, damit diese Funktion nichts ueber den Einstellungsspeicher wissen muss. Rueckgabe: was tatsaechlich uebernommen wurde -- der Aufrufer soll es benennen koennen. """ alt = read_vendor_settings(legacy) uebernommen: dict[str, str] = {} for key, wert in alt.items(): if str(read(key) or "").strip(): continue # was hier schon steht, gilt write(key, wert) uebernommen[key] = wert return uebernommen def import_all( legacy: sqlite3.Connection, conn: sqlite3.Connection, *, keystore: Keystore, master_private_key: str, master_public_key: str, master_endpoint: str, vendor: dict[str, Any], actor: str = "migration", reissue: bool = True, ) -> Report: """Uebernimmt Kunden, Tickets und Lizenzen. Erwartet eine offene Transaktion.""" report = Report() # -- Kunden ------------------------------------------------------------ # customer_map: dict[int, int] = {} for row in legacy.execute("SELECT * FROM license_customers ORDER BY id"): name = str(row["name"] or "").strip() if not name: report.skipped.append(f"Kunde ohne Namen (alte Nummer {row['id']})") continue existing = Database.one( conn, "SELECT id FROM license_customers WHERE name=? COLLATE NOCASE", (name,) ) if existing is not None: customer_map[int(row["id"])] = int(existing["id"]) report.warnings.append(f"Kunde {name!r} war schon vorhanden -- nicht doppelt angelegt.") continue new_id = issuance.create_customer( conn, actor=actor, name=name, contact_email=str(row["contact_email"] or ""), contact_phone=str(row["contact_phone"] or ""), contact_person=str(row["contact_person"] or ""), street=str(row["street"] or ""), house_number=str(row["house_number"] or ""), postal_code=str(row["postal_code"] or ""), city=str(row["city"] or ""), notes=str(row["notes"] or ""), ) customer_map[int(row["id"])] = new_id report.customers += 1 # -- Tickets ----------------------------------------------------------- # for row in legacy.execute("SELECT * FROM license_tickets ORDER BY created_at"): customer_id = customer_map.get(int(row["customer_id"] or 0)) if customer_id is None: report.skipped.append(f"Ticket {str(row['id'])[:12]} -- Kunde nicht uebernommen") continue if Database.one(conn, "SELECT id FROM license_tickets WHERE id=?", (row["id"],)): continue conn.execute( "INSERT INTO license_tickets (id, customer_id, type, modules, valid_days, lifetime, " "status, role, notes, created_by, created_at, updated_at) " "VALUES (?,?,?,?,?,?,?,?,?,?,?,?)", ( str(row["id"]), customer_id, str(row["type"] or "standard"), json.dumps(_modules(row["modules"])), int(row["valid_days"] or 365), int(row["lifetime"] or 0), str(row["status"] or "open"), str(row["role"] or "customer"), "Aus der Vorgaengerfassung uebernommen.", str(row["created_by"] or actor), str(row["created_at"] or _now()), _now(), ), ) report.tickets += 1 # -- Lizenzen ---------------------------------------------------------- # for row in legacy.execute("SELECT * FROM licenses ORDER BY created_at"): license_id = str(row["license_id"] or "") customer_id = customer_map.get(int(row["customer_id"] or 0)) if not license_id or customer_id is None: report.skipped.append(f"Lizenz {license_id[:12] or '?'} -- Kunde nicht uebernommen") continue if Database.one(conn, "SELECT id FROM licenses WHERE license_id=?", (license_id,)): continue customer = issuance.get_customer(conn, customer_id) status = str(row["status"] or "issued") modules = _modules(row["modules"]) if not reissue: report.skipped.append(f"{customer['name']} / {license_id[:12]} -- nur gezaehlt") report.licenses += 1 continue # Neu ausstellen: gleiche Kennung, gleiche Eckdaten, neues Format und # ein frisches Clientschluesselpaar. Das alte Paar gehoert zu einer # Signatur, die es nicht mehr gibt. lifetime = bool(row["lifetime"]) valid_days = _remaining_days(str(row["expires_at"] or "")) if not lifetime and valid_days <= 0: report.skipped.append( f"{customer['name']} / {license_id[:12]} -- abgelaufen am " f"{str(row['expires_at'])[:10]}, nicht uebernommen" ) continue bundle, license_public = licsrv.issue_license( customer={ "name": str(customer["name"]), "contact_email": str(customer["contact_email"] or ""), "customer_ref": str(customer["id"]), }, license_type=str(row["type"] or "standard"), modules=modules, valid_days=valid_days, master_private_key_b64=master_private_key, master_public_key_b64=master_public_key, master_endpoint=master_endpoint, vendor=vendor, license_id=license_id, role=str(row["role"] or "customer"), lifetime=lifetime, ) document = bundle["license"] conn.execute( "INSERT INTO licenses (license_id, ticket_id, customer_id, type, role, modules, " "lifetime, seq, issued_at, expires_at, license_pubkey, license_document, " "client_key_encrypted, status, created_by, created_at) " "VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)", ( license_id, str(row["ticket_id"] or ""), customer_id, str(row["type"] or "standard"), str(row["role"] or "customer"), json.dumps(modules), int(lifetime), int(document["seq"]), document["issued_at"], document["expires_at"], license_public, json.dumps(document), keystore.encrypt( str(bundle["client_key"]), aad=issuance.client_key_aad(license_id) ), # Kein Zustand, der eine Maschine nennt: die alte Bindung gilt # fuer die alte Datei, nicht fuer diese. "revoked" if status == "revoked" else "issued", str(row["created_by"] or actor), str(row["created_at"] or _now()), ), ) report.licenses += 1 if status != "revoked": report.reissued.append( f"{customer['name']} -- {license_id} " f"({'aktiv gewesen' if status == 'active' else status})" ) if report.reissued: report.warnings.append( "Die neu signierten Lizenzen sind noch nicht aktiviert: die Bindung an eine " "Maschine gilt nur fuer die alte Datei. Nach dem Einspielen aktiviert der " "Kunde erneut." ) return report