ssh-jumphost/scripts/diff_effective_rights.py

#!/usr/bin/env python3
"""D.7-Pflichtcheck (Umsetzungsauftrag_Sonnet5.md, Teil D): "Vorher/Nachher-
Diff der effektiven Rechte je Benutzer" als Migrationspruefung.

Berechnet fuer jeden aktiven (nicht soft-geloeschten) Benutzer und jede
bekannte Rolle (dynamisch aus app.models.schemas.ROLE_NAME gelesen, damit
das Skript nicht bei jeder Rollenaenderung von Hand nachgezogen werden
muss) die Menge der Hostgruppen-IDs, fuer die die Rolle effektiv gilt --
ueber exakt dieselbe Logik wie die Anwendung selbst
(app.rbac.user_host_group_ids_with_any_role, die einzige Wahrheitsquelle
fuer RBAC-Aufloesung, siehe FORTSETZUNG_Teil_D.md Abschnitt 1, S13).

Zwei Betriebsarten:

1. Snapshot exportieren (vor UND nach einer Migration je einmal aufrufen):
       <venv>/bin/python scripts/diff_effective_rights.py \\
           --dump /pfad/zur.db --out vorher.json

2. Zwei bestehende SQLite-Dateien direkt gegeneinander diffen (z.B. eine
   Kopie der Produktions-DB vor dem Wartungsfenster gegen die DB danach):
       <venv>/bin/python scripts/diff_effective_rights.py \\
           --before vorher.db --after nachher.db

   Exit-Code 0: identische effektive Rechte (erwartetes Ergebnis fuer eine
   rechteneutrale Migration wie D.4/D.6-Schritt-3). Exit-Code 1: mindestens
   eine Abweichung gefunden -- Details werden ausgegeben, das Wartungsfenster
   sollte dann NICHT fortgesetzt werden, ohne die Abweichung verstanden zu
   haben.

Kann beliebig oft und gefahrlos gegen eine DB-Kopie aufgerufen werden --
rein lesend (oeffnet SQLite explizit read-only ueber die URI-Form
`file:...?mode=ro`), schreibt nichts.
"""
from __future__ import annotations

import argparse
import asyncio
import json
import sys
import typing
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent.parent))

import aiosqlite  # noqa: E402

from app.models.schemas import ROLE_NAME  # noqa: E402
from app.rbac import user_host_group_ids_with_any_role  # noqa: E402

ROLES: tuple[str, ...] = typing.get_args(ROLE_NAME)


async def snapshot(db_path: Path) -> dict[str, dict]:
    """Liest den DB-Pfad read-only und liefert
    {user_id_str: {"username": ..., "rights": {role_name: [host_group_id, ...]}}}
    -- nur Rollen mit mindestens einer Hostgruppe werden aufgefuehrt, damit
    der Diff bei Uebereinstimmung leer/kompakt bleibt."""
    uri = f"file:{db_path}?mode=ro"
    async with aiosqlite.connect(uri, uri=True) as conn:
        cursor = await conn.execute(
            "SELECT id, username FROM users WHERE deleted_at IS NULL ORDER BY id"
        )
        users = await cursor.fetchall()

        result: dict[str, dict] = {}
        for user_id, username in users:
            rights: dict[str, list[int]] = {}
            for role in ROLES:
                host_group_ids = await user_host_group_ids_with_any_role(
                    conn, user_id=user_id, role_names=(role,)
                )
                if host_group_ids:
                    rights[role] = sorted(host_group_ids)
            result[str(user_id)] = {"username": username, "rights": rights}
        return result


def diff_snapshots(before: dict[str, dict], after: dict[str, dict]) -> list[str]:
    """Liefert eine Liste menschenlesbarer Abweichungszeilen; leere Liste ==
    identische effektive Rechte."""
    lines: list[str] = []
    all_ids = sorted(set(before) | set(after), key=lambda x: int(x))

    for uid in all_ids:
        b = before.get(uid)
        a = after.get(uid)
        if b is None:
            lines.append(f"user_id={uid} ({a['username']!r}): NEU nach der Migration (vorher nicht aktiv/vorhanden)")
            continue
        if a is None:
            lines.append(f"user_id={uid} ({b['username']!r}): VERSCHWUNDEN nach der Migration (vorher aktiv)")
            continue

        b_rights, a_rights = b["rights"], a["rights"]
        roles = sorted(set(b_rights) | set(a_rights))
        for role in roles:
            b_set = set(b_rights.get(role, []))
            a_set = set(a_rights.get(role, []))
            if b_set != a_set:
                gained = sorted(a_set - b_set)
                lost = sorted(b_set - a_set)
                detail = []
                if gained:
                    detail.append(f"NEU auf Hostgruppen {gained}")
                if lost:
                    detail.append(f"VERLOREN auf Hostgruppen {lost}")
                lines.append(
                    f"user_id={uid} ({b['username']!r}), Rolle={role!r}: "
                    + "; ".join(detail)
                )
    return lines


async def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    parser.add_argument("--dump", type=Path, help="DB-Datei, deren Rechte-Snapshot exportiert werden soll")
    parser.add_argument("--out", type=Path, help="Zieldatei fuer --dump (JSON)")
    parser.add_argument("--before", type=Path, help="DB-Datei VOR der Migration")
    parser.add_argument("--after", type=Path, help="DB-Datei NACH der Migration")
    args = parser.parse_args()

    if args.dump:
        if not args.out:
            parser.error("--dump erfordert --out")
        data = await snapshot(args.dump)
        args.out.write_text(json.dumps(data, indent=2, sort_keys=True, ensure_ascii=False), encoding="utf-8")
        print(f"Snapshot von {args.dump} -> {args.out} geschrieben ({len(data)} aktive Benutzer).")
        return 0

    if args.before and args.after:
        before = await snapshot(args.before)
        after = await snapshot(args.after)
        diffs = diff_snapshots(before, after)
        if not diffs:
            print(f"KEINE Abweichung: effektive Rechte fuer {len(before)} Benutzer identisch vor/nach der Migration.")
            return 0
        print(f"{len(diffs)} Abweichung(en) gefunden:")
        for line in diffs:
            print(f"  - {line}")
        return 1

    parser.error("entweder --dump zusammen mit --out, oder --before zusammen mit --after angeben")
    return 2


if __name__ == "__main__":
    raise SystemExit(asyncio.run(main()))