ssh-jumphost/scripts/pre_schritt4_checks.py

#!/usr/bin/env python3
"""Pflicht-Vorab-Reports vor Teil D Schritt 4 (Umsetzungsauftrag_Sonnet5.md,
D.6 Schritt 4 und D.7-Risikotabelle: "Zugriffsausfall durch Achse B" und
"Verhaltensaenderung bei der Credential-Auswahl").

Zwei unabhaengige Reports gegen eine SQLite-Datei (Produktions-DB-Kopie oder
die aktuelle Dev-DB):

1. "Hosts mit Zugangsdaten ohne Gruppenfreigabe" -- ein Host hat einen SSH-
   Schluessel, ein SSH-Passwort oder einen RDP-Zugangsdatensatz zugeordnet
   (host_*_map), aber KEINE Benutzergruppe hat diesen Datensatz ueber Achse
   B (group_*_grants, Migration 0018) freigegeben. Nach Schritt 4 kann sich
   niemand mehr darueber anmelden -- das MUSS vor Schritt 4 bereinigt sein
   (z.B. durch eine nachtraegliche manuelle Achse-B-Freigabe), sonst ist der
   Host fuer alle Benutzer unerreichbar, ungeachtet ihrer ssh_connect/
   rdp_connect-Rolle.

2. "Hosts mit mehrdeutigen SSH-Keys" -- mehr als ein Eintrag in
   host_ssh_key_map fuer denselben Host. Strukturell nur bei SSH-Keys
   moeglich (n:m-Zuordnung); RDP- und SSH-Passwort-Zuordnung sind laut
   Schema 1:1 (PK auf host_id). Ab Schritt 4 wirft
   app.rbac.resolve_credential_for_user_on_host() einen harten
   AmbiguousCredentialError, sobald ein Benutzer ueber mehr als eine seiner
   Gruppen Zugriff auf mehr als einen der zugeordneten Schluessel hat --
   dieser Report zeigt, wo das ueberhaupt strukturell moeglich ist, damit
   vorab geklaert werden kann, ob das gewollt ist (z.B. Umbenennen/
   Entfernen des ueberzaehligen Schluessels).

Aufruf:
    <venv>/bin/python scripts/pre_schritt4_checks.py --db /pfad/zur.db

Exit-Code 0: beide Reports leer (unbedenklich, Schritt 4 kann bedenkenlos
scharf geschaltet werden). Exit-Code 1: mindestens ein Fund in einem der
beiden Reports -- Details werden ausgegeben.

Rein lesend (SQLite read-only via `file:...?mode=ro`), schreibt nichts.
"""
from __future__ import annotations

import argparse
import asyncio
import sys
from pathlib import Path

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

import aiosqlite  # noqa: E402

# (kind, host_map_table, host_map_credential_col, grant_table, grant_credential_col)
_CHECKS = [
    ("ssh_key", "host_ssh_key_map", "ssh_key_id", "group_ssh_key_grants", "ssh_key_id"),
    (
        "rdp_credential",
        "host_rdp_credential_map",
        "rdp_credential_id",
        "group_rdp_credential_grants",
        "rdp_credential_id",
    ),
    (
        "ssh_password_credential",
        "host_ssh_password_credential_map",
        "ssh_password_credential_id",
        "group_ssh_password_credential_grants",
        "ssh_password_credential_id",
    ),
]


async def report_orphaned_credential_mappings(conn: aiosqlite.Connection) -> list[str]:
    """Report 1: host_*_map-Zeilen, deren Credential-ID in keiner Zeile der
    zugehoerigen group_*_grants-Tabelle vorkommt (unabhaengig von expires_at
    -- ein abgelaufener Grant zaehlt hier bewusst NICHT als Deckung)."""
    lines: list[str] = []
    for kind, map_table, map_col, grant_table, grant_col in _CHECKS:
        cursor = await conn.execute(
            f"""
            SELECT m.host_id, h.hostname, m.{map_col}
            FROM {map_table} m
            JOIN hosts h ON h.id = m.host_id
            WHERE NOT EXISTS (
                SELECT 1 FROM {grant_table} g
                WHERE g.{grant_col} = m.{map_col}
                  AND (g.expires_at IS NULL OR g.expires_at > strftime('%Y-%m-%dT%H:%M:%fZ','now'))
            )
            ORDER BY m.host_id
            """
        )
        rows = await cursor.fetchall()
        for host_id, hostname, credential_id in rows:
            lines.append(
                f"kind={kind}: Host {host_id} ({hostname!r}) hat {kind}={credential_id} "
                "zugeordnet, aber KEINE Gruppe hat ihn ueber Achse B freigegeben."
            )
    return lines


async def report_ambiguous_ssh_keys(conn: aiosqlite.Connection) -> list[str]:
    """Report 2: Hosts mit mehr als einem Eintrag in host_ssh_key_map."""
    cursor = await conn.execute(
        """
        SELECT m.host_id, h.hostname, COUNT(*) AS n, GROUP_CONCAT(m.ssh_key_id)
        FROM host_ssh_key_map m
        JOIN hosts h ON h.id = m.host_id
        GROUP BY m.host_id
        HAVING COUNT(*) > 1
        ORDER BY m.host_id
        """
    )
    rows = await cursor.fetchall()
    return [
        f"Host {host_id} ({hostname!r}): {n} SSH-Keys zugeordnet (ssh_key_ids={key_ids}) "
        "-- Mehrdeutigkeit strukturell moeglich, sobald mehr als einer davon derselben "
        "Benutzergruppe freigegeben wird."
        for host_id, hostname, n, key_ids in rows
    ]


async def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    parser.add_argument("--db", type=Path, required=True, help="Pfad zur zu pruefenden SQLite-Datei")
    args = parser.parse_args()

    uri = f"file:{args.db}?mode=ro"
    async with aiosqlite.connect(uri, uri=True) as conn:
        orphaned = await report_orphaned_credential_mappings(conn)
        ambiguous = await report_ambiguous_ssh_keys(conn)

    print("=== Report 1: Hosts mit Zugangsdaten ohne Gruppenfreigabe ===")
    if not orphaned:
        print("  Keine Funde.")
    else:
        for line in orphaned:
            print(f"  - {line}")

    print()
    print("=== Report 2: Hosts mit mehrdeutigen SSH-Keys ===")
    if not ambiguous:
        print("  Keine Funde.")
    else:
        for line in ambiguous:
            print(f"  - {line}")

    return 1 if (orphaned or ambiguous) else 0


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