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:
    return os.path.ismount(_mountpoint(name))


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 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"]
    ergebnis = subprocess.run(cmd, capture_output=True, text=True, timeout=20)
    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(f"Freigaben-Abfrage fehlgeschlagen: {fehler[:200]}")
    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()
            raise RuntimeError(f"mount schlug fehl: {fehler[:300]}")
        return schreibtest(ziel)
    finally:
        if creds_datei:
            try:
                os.unlink(creds_datei)
            except OSError:
                pass


def aushaengen(name: str) -> None:
    ziel = _mountpoint(name)
    if os.path.ismount(ziel):
        ergebnis = subprocess.run(
            ["umount", ziel], capture_output=True, text=True, timeout=30
        )
        if ergebnis.returncode != 0:
            raise RuntimeError(
                f"umount schlug fehl: {(ergebnis.stderr or '').strip()[:300]}"
            )
    try:
        os.rmdir(ziel)
    except OSError:
        pass  # nicht leer oder weg — egal


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