rippy/docker/api/mounts.py

"""Netzwerk-Speicherziele (NFS/SMB) direkt aus Rippy heraus einhängen.

Commander-Anforderung 23.07.: Data-Mounts über das UI, universell für jede
Installation. Der api-Container läuft dafür mit CAP_SYS_ADMIN und einem
rshared-Bind auf /app/media — ein Mount hier propagiert über den Host in
alle anderen Container (Worker sieht das Ziel sofort).

Zugangsdaten liegen in Postgres (Klartext — bewusster Heimnetz-Kompromiss,
im README dokumentiert; SMB-Credentials wandern NIE in die Kommandozeile,
sondern über eine temporäre credentials-Datei).
"""

import os
import re
import subprocess
import tempfile

import db

MEDIA_ROOT = "/app/media"
NAME_MUSTER = re.compile(r"^[a-z0-9][a-z0-9-]{1,30}$")


def _mountpoint(name: str) -> str:
    return os.path.join(MEDIA_ROOT, name)


def validiere_name(name: str) -> bool:
    return bool(NAME_MUSTER.match(name))


def ist_gemountet(name: str) -> bool:
    """Ist unter /app/media/<name> etwas gemountet? Ein TOTER CIFS-Mount
    (NAS weg) lässt os.path.ismount mit OSError fliegen — das zählt weiter
    als 'gemountet' (nur eben kaputt), damit die Reparatur greift."""
    try:
        return os.path.ismount(_mountpoint(name))
    except OSError:
        return True


def ist_erreichbar(name: str) -> bool:
    """Kann auf das Ziel WIRKLICH zugegriffen werden?

    Ein toter CIFS-Mount (Host weg, stale) ist zwar 'gemountet', aber jeder
    Zugriff scheitert mit „Host is down" (Befund 24.07.). CIFS läuft mit
    `soft`, d. h. der Zugriff bricht schnell ab statt zu hängen.
    """
    try:
        os.listdir(_mountpoint(name))
        return True
    except OSError:
        return False


def schreibtest(pfad: str) -> bool:
    """Berechtigungs-Prüfung: können wir im Ziel wirklich schreiben?

    Commander-Anforderung 23.07.: beim Hinzufügen muss geprüft werden, ob
    die Rechte stimmen — ein nur-lesbares Ziel fällt sonst erst beim Rip auf.
    """
    testdatei = os.path.join(pfad, ".rippy-schreibtest")
    try:
        with open(testdatei, "w") as f:
            f.write("ok")
        os.unlink(testdatei)
        return True
    except OSError:
        return False


def uebersetze_smb_fehler(fehler: str, mit_credentials: bool) -> str:
    """Pure Funktion (testbar): NT_STATUS-Kauderwelsch → handelbarer Klartext.

    Befund 24.07.: „session setup failed: NT_STATUS_ACCESS_DENIED" hieß in
    Wahrheit nur „Windows/NAS erlauben keine Gast-Abfrage" — ohne Übersetzung
    rät niemand, dass oben Benutzer/Passwort fehlen.
    """
    if "NT_STATUS_ACCESS_DENIED" in fehler:
        if not mit_credentials:
            return (
                "Zugriff verweigert — dieser Rechner erlaubt keine Gast-Abfrage. "
                "Benutzername + Passwort eintragen (Windows: ein Konto mit "
                "Zugriff auf die Freigabe), dann erneut auflisten."
            )
        return (
            "Zugriff verweigert — Benutzername/Passwort stimmen nicht oder das "
            "Konto darf die Freigaben nicht auflisten."
        )
    if "NT_STATUS_LOGON_FAILURE" in fehler:
        return "Anmeldung fehlgeschlagen — Benutzername oder Passwort falsch."
    if (
        "NT_STATUS_HOST_UNREACHABLE" in fehler
        or "NT_STATUS_IO_TIMEOUT" in fehler
        or "NT_STATUS_UNSUCCESSFUL" in fehler
        or "NT_STATUS_CONNECTION_REFUSED" in fehler
        or "Connection to" in fehler
    ):
        return (
            "Rechner nicht erreichbar — Name/IP prüfen; läuft dort ein "
            "SMB-Dienst (Port 445, Datei- und Druckerfreigabe aktiv)?"
        )
    return fehler[:200]


def liste_smb_freigaben(host: str, username: str = "", passwort: str = "") -> list:
    """Listet die SMB-Freigaben eines Rechners (smbclient -L) — damit man
    seinen PC/NAS per Klick wählt statt //host/share zu raten."""
    cmd = ["smbclient", "-L", f"//{host}", "-g"]
    if username:
        cmd += ["-U", f"{username}%{passwort or ''}"]
    else:
        cmd += ["-N"]
    try:
        ergebnis = subprocess.run(cmd, capture_output=True, text=True, timeout=20)
    except subprocess.TimeoutExpired:
        raise RuntimeError(
            f"Freigaben-Abfrage fehlgeschlagen: {host} antwortet nicht "
            "(Timeout nach 20 s) — Name/IP und Netzwerk prüfen."
        )
    freigaben = []
    for zeile in (ergebnis.stdout or "").splitlines():
        # -g-Format: Disk|Freigabename|Kommentar
        teile = zeile.split("|")
        if len(teile) >= 2 and teile[0] == "Disk" and not teile[1].endswith("$"):
            freigaben.append(teile[1])
    if not freigaben and ergebnis.returncode != 0:
        fehler = (ergebnis.stderr or ergebnis.stdout or "").strip()
        raise RuntimeError(
            "Freigaben-Abfrage fehlgeschlagen: "
            + uebersetze_smb_fehler(fehler, bool(username))
        )
    return freigaben


def mounten(name: str, typ: str, quelle: str, optionen: str = "",
            username: str = "", passwort: str = "") -> bool:
    """Hängt ein NFS/CIFS-Ziel unter /app/media/<name> ein.

    Rückgabe: True wenn das Ziel SCHREIBBAR ist. Wirft RuntimeError bei
    Mount-Fehlern.
    """
    ziel = _mountpoint(name)
    os.makedirs(ziel, exist_ok=True)
    if os.path.ismount(ziel):
        return schreibtest(ziel)

    creds_datei = None
    try:
        if typ == "nfs":
            opts = optionen or "vers=4,soft,timeo=100"
            cmd = ["mount", "-t", "nfs", "-o", opts, quelle, ziel]
        elif typ == "cifs":
            teile = [optionen] if optionen else []
            if username:
                # Credentials über Datei statt Kommandozeile (ps-sichtbar!)
                creds = tempfile.NamedTemporaryFile(
                    "w", delete=False, prefix="cifs-", suffix=".cred"
                )
                creds.write(f"username={username}\npassword={passwort or ''}\n")
                creds.close()
                os.chmod(creds.name, 0o600)
                creds_datei = creds.name
                teile.append(f"credentials={creds.name}")
            else:
                teile.append("guest")
            teile.append("iocharset=utf8")
            cmd = ["mount", "-t", "cifs", "-o", ",".join(teile), quelle, ziel]
        else:
            raise RuntimeError(f"Unbekannter Typ: {typ} (nfs oder cifs)")

        ergebnis = subprocess.run(cmd, capture_output=True, text=True, timeout=30)
        if ergebnis.returncode != 0:
            fehler = (ergebnis.stderr or ergebnis.stdout or "").strip()
            hinweis = ""
            if "error(13)" in fehler or "Permission denied" in fehler:
                hinweis = (
                    " — Zugriff verweigert: Benutzer/Passwort prüfen; ohne "
                    "Angaben versucht Rippy einen Gast-Zugriff, den Windows/"
                    "NAS meist ablehnen."
                )
            raise RuntimeError(f"mount schlug fehl: {fehler[:300]}{hinweis}")
        return schreibtest(ziel)
    finally:
        if creds_datei:
            try:
                os.unlink(creds_datei)
            except OSError:
                pass


def _lazy_umount(ziel: str) -> None:
    """`umount -l`: hängt auch einen TOTEN/beschäftigten Mount ab (detach now,
    cleanup later). Ohne das ließ sich ein Mount zu einem weggefallenen NAS
    gar nicht mehr entfernen (Befund 24.07.)."""
    subprocess.run(["umount", "-l", ziel], capture_output=True, text=True, timeout=30)


def aushaengen(name: str) -> None:
    ziel = _mountpoint(name)
    try:
        noch_mount = os.path.ismount(ziel)
    except OSError:
        noch_mount = True  # stale
    if noch_mount:
        ergebnis = subprocess.run(
            ["umount", ziel], capture_output=True, text=True, timeout=30
        )
        if ergebnis.returncode != 0:
            # Toter/beschäftigter Mount → lazy detach (klappt immer)
            _lazy_umount(ziel)
    try:
        os.rmdir(ziel)
    except OSError:
        pass  # nicht leer oder weg — egal


def reparieren(name: str, typ: str, quelle: str, optionen: str = "",
               username: str = "", passwort: str = "") -> bool:
    """Toten/veralteten Mount frisch neu verbinden: lazy abhängen, neu mounten.

    Nötig, wenn ein NAS-Mount stale geworden ist (Rebuild, NAS-Schlaf) — das
    normale mounten() würde am „ist schon Mountpoint" hängenbleiben.
    """
    ziel = _mountpoint(name)
    try:
        if os.path.ismount(ziel):
            _lazy_umount(ziel)
    except OSError:
        _lazy_umount(ziel)
    return mounten(name, typ, quelle, optionen, username, passwort)


def alle_remounten() -> list:
    """Beim API-Start: alle gespeicherten Mounts wiederherstellen."""
    meldungen = []
    for eintrag in db.list_mounts():
        try:
            mounten(
                eintrag["name"], eintrag["typ"], eintrag["quelle"],
                eintrag.get("optionen") or "", eintrag.get("username") or "",
                eintrag.get("passwort") or "",
            )
            meldungen.append(f"{eintrag['name']}: eingehängt")
        except Exception as e:
            meldungen.append(f"{eintrag['name']}: FEHLER — {e}")
    return meldungen