tesm-license/deploy/tesm-motd

#!/usr/bin/env python3
"""Live-Banner fuer die SSH-Anmeldung (MOTD).

Eingebunden ueber ``/etc/update-motd.d/50-<name>``, das ``install.sh``
erzeugt. Aufruf mit ``--app`` und ``--root``; alles andere steht in der
Datenbank der Installation.

**Nur Standardbibliothek, kein Flask, kein venv.** Der Vorgaenger importierte
dafuer die ganze Anwendung -- bei jeder SSH-Anmeldung. Das ist teuer, und es
faellt aus, sobald das venv nicht mehr passt: nach dem Umstieg auf diese
Fassung zeigte das alte Banner monatelang eingefrorene Zahlen aus einer
Datenbank, die niemand mehr fuellte. Hier wird gelesen, was auf der Platte
liegt, und zwar **nur lesend** (``mode=ro``).

Ein Banner darf eine Anmeldung niemals stoeren: jede Auskunft, die nicht zu
holen ist, wird zu einem kurzen Hinweis, und am Ende faengt ``main`` alles ab.
Der Aufrufer setzt zusaetzlich ein Zeitlimit.
"""

from __future__ import annotations

import argparse
import sqlite3
import sys
import time
from pathlib import Path

AMBER = "\033[38;2;226;166;60m"
DIM = "\033[2m"
BOLD = "\033[1m"
RESET = "\033[0m"

#: Kea schreibt seine Leases als CSV; die letzte Zeile je Adresse gilt.
#: Dieselbe Datei liest die Anwendung (``services/kea.py``: ``LEASE_FILE``).
LEASE_FILE = "/var/lib/kea/kea-leases4.csv"

#: Zeile unter der Wortmarke. Wortlaut wie in der jeweiligen Anwendung
#: (``AUTH_TAGLINE`` in deren ``__init__.py``) -- ein Hygienetest haelt beides
#: zusammen.
TAGLINE = {
    "tesm": "ÜBERWACHEN · BOOTEN · ANBINDEN",
    "tesm-license": "LIZENZIEREN",
}

TITEL = {"tesm": "TimEShepManager", "tesm-license": "Lizenzserver"}

#: Wortgleich mit ``services/monitor.py``: ``DASHBOARD_COUNT_SQL``. Die
#: Statusuebersicht und dieses Banner sollen nicht zwei Wahrheiten haben; ein
#: Hygienetest vergleicht die beiden Zeichenketten.
DASHBOARD_COUNT_SQL = """
        SELECT
          COUNT(*) AS total,
          SUM(CASE WHEN d.is_active=0 THEN 1 ELSE 0 END) AS disabled,
          SUM(CASE WHEN d.is_active=1 AND COALESCE(st.state,'unknown')='online' THEN 1 ELSE 0 END) AS online,
          SUM(CASE WHEN d.is_active=1 AND COALESCE(st.state,'unknown')='offline' THEN 1 ELSE 0 END) AS offline,
          SUM(CASE WHEN d.is_active=1 AND COALESCE(st.state,'unknown')='unknown' THEN 1 ELSE 0 END) AS unknown,
          SUM(CASE WHEN d.is_active=1 AND st.state='restarting' THEN 1 ELSE 0 END) AS restarting
        FROM devices d LEFT JOIN device_status st ON st.device_id = d.id
        WHERE d.deleted_at IS NULL
        """


def _zahl(anzahl: int, singular: str, plural: str) -> str:
    """Zahl mit passendem Wort. "1 Clients" liest man jeden Tag einmal zu oft."""
    return f"{BOLD}{anzahl}{RESET} {singular if anzahl == 1 else plural}"


def _oeffnen(root: Path) -> sqlite3.Connection:
    pfad = root / "data" / "app.db"
    conn = sqlite3.connect(f"file:{pfad.as_posix()}?mode=ro", uri=True, timeout=1.0)
    conn.row_factory = sqlite3.Row
    return conn


def _flotte(conn: sqlite3.Connection) -> str:
    row = conn.execute(DASHBOARD_COUNT_SQL).fetchone()
    online = int(row["online"] or 0)
    offline = int(row["offline"] or 0)
    teile = [
        _zahl(online, "Client online", "Clients online"),
        f"{BOLD}{offline}{RESET} nicht erreichbar",
    ]
    # Nur nennen, wenn es sie gibt: eine Null bei "startet neu" ist keine
    # Auskunft, sondern Fuellmaterial.
    restarting = int(row["restarting"] or 0)
    if restarting:
        teile.append(_zahl(restarting, "startet neu", "starten neu"))
    unbekannt = int(row["unknown"] or 0)
    if unbekannt:
        teile.append(f"{DIM}{unbekannt} ohne Pruefung{RESET}")
    return " · ".join(teile)


def _dhcp() -> str:
    """Aktive Leases aus der Kea-CSV. Laeuft als root, braucht keinen Helfer."""
    datei = Path(LEASE_FILE)
    if not datei.is_file():
        return "keine Lease-Datei"
    try:
        zeilen = datei.read_text(encoding="utf-8", errors="replace").splitlines()
    except OSError:
        return "Lease-Datei nicht lesbar"
    if len(zeilen) < 2:
        return "0 Leases aktiv"
    kopf = [name.strip().strip('"') for name in zeilen[0].split(",")]
    try:
        i_adresse = kopf.index("address")
        i_ablauf = kopf.index("expire")
        i_zustand = kopf.index("state")
    except ValueError:
        return "Lease-Datei mit unerwarteten Spalten"

    jetzt = int(time.time())
    letzte: dict[str, tuple[int, str]] = {}
    for zeile in zeilen[1:]:
        felder = [feld.strip().strip('"') for feld in zeile.split(",")]
        if len(felder) < len(kopf):
            continue
        try:
            letzte[felder[i_adresse]] = (int(felder[i_ablauf] or 0), felder[i_zustand] or "0")
        except ValueError:
            continue
    aktiv = sum(1 for ablauf, zustand in letzte.values() if zustand == "0" and ablauf > jetzt)
    return f"{aktiv} {'Lease' if aktiv == 1 else 'Leases'} aktiv"


def _lizenzen(conn: sqlite3.Connection) -> str:
    kunden = int(conn.execute("SELECT COUNT(*) FROM license_customers").fetchone()[0] or 0)
    lizenzen = int(
        conn.execute(
            "SELECT COUNT(*) FROM licenses WHERE status IN ('issued','active')"
        ).fetchone()[0]
        or 0
    )
    return (
        _zahl(kunden, "Kunde", "Kunden")
        + " · "
        + _zahl(lizenzen, "Lizenz ausgestellt", "Lizenzen ausgestellt")
    )


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(prog="tesm-motd", description="Banner fuer die SSH-Anmeldung")
    parser.add_argument("--app", required=True, choices=sorted(TITEL))
    parser.add_argument("--root", required=True, help="Verzeichnis der Installation")
    parser.add_argument("--no-color", action="store_true")
    args = parser.parse_args(argv)

    if args.no_color:
        globals().update(AMBER="", DIM="", BOLD="", RESET="")

    root = Path(args.root)
    # Kein isatty()-Test: dieses Skript schreibt beim Login nicht auf ein
    # Terminal, sondern ueber run-parts in eine Datei. Der Betrachter ist
    # trotzdem ein echtes Terminal -- deshalb immer Farbe.
    print(f"{BOLD}{AMBER}TESM{RESET} · {TITEL[args.app]}")
    if TAGLINE[args.app]:
        print(f"{DIM}{TAGLINE[args.app]}{RESET}")
    print()
    try:
        conn = _oeffnen(root)
    except sqlite3.Error:
        print(f"{DIM}Datenbank nicht lesbar: {root}{RESET}")
        print()
        return 0
    try:
        if args.app == "tesm":
            print(f"{_flotte(conn)} · DHCP: {_dhcp()}")
        else:
            print(_lizenzen(conn))
    except sqlite3.Error as fehler:
        print(f"{DIM}Zustand nicht verfuegbar ({fehler}){RESET}")
    finally:
        conn.close()
    print()
    return 0


if __name__ == "__main__":
    try:
        raise SystemExit(main())
    except SystemExit:
        raise
    except Exception:  # noqa: BLE001 - ein Banner darf keinen Login stoeren
        sys.exit(0)