feat: Anmeldung abschaltbar (vertraute Netze oder ganz); Doku 0.5.0

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
feat: gespeicherte Zugangsdaten für SMB und SFTP-Passwörter
2026-10-11 01:12:28 +02:00 · 2026-10-11 01:07:09 +02:00 · 2026-10-11 01:00:30 +02:00 · 2026-10-11 01:00:30 +02:00
36 changed files with 1720 additions and 213 deletions
@@ -8,6 +8,27 @@ Die Version steht an genau einer Stelle: `src/netmover/__init__.py`.

 ## [Unreleased]

+## [0.5.0] – 2026-10-11
+
+### Hinzugefügt
+- **SFTP als Quelle** mit denselben Garantien wie bei SMB/NFS: Stabilitätsprüfung,
+  atomares Schreiben, Löschen der Quelle erst nach vollständiger Übertragung,
+  Kopier-Buchführung, Quarantäne. Verbindungsabbrüche brechen den Lauf als gestörte
+  Anbindung ab. Bei SFTP-Quellen mindestens 2 s Stabilitätsfenster (Zeiten nur
+  sekundengenau).
+- **Gespeicherte Zugangsdaten** für SMB und SFTP-Passwörter (Benutzer, Passwort,
+  Domäne); in Anbindungen auswählbar, ein neues Passwort gilt sofort für alle.
+  Die Seite „Schlüssel“ heißt jetzt **„Zugangsdaten“** (Passwörter + SSH-Schlüssel);
+  `/keys` leitet weiter.
+- **Anmeldung abschaltbar** (*Einstellungen → Anmeldung*, Bestätigung mit Passwort):
+  immer / nicht aus vertrauenswürdigen Netzen / nie (z. B. hinter Reverse Proxy mit
+  eigener Anmeldung). Deutlicher Hinweis in der Kopfzeile; CSRF-Schutz bleibt;
+  `netmover reset-password` schaltet die Anmeldung wieder ein.
+
+### Geändert
+- Mover intern auf eine Quell-Schnittstelle umgestellt (`LocalSource`, `SftpSource`);
+  bestehendes Verhalten unverändert (alle bisherigen Tests gelten weiter).
+
 ## [0.4.0] – 2026-10-11

 ### Hinzugefügt
@@ -213,7 +234,8 @@ Erste Version.
 - Löschen von Zugangsdaten, die noch von einem Mount benutzt werden, trennt den
  Mount nicht mehr stillschweigend ab, sondern wird von der Datenbank abgelehnt.

-[Unreleased]: https://git.wilde-server.de/florian.wilde/netmover/compare/v0.4.0...HEAD
+[Unreleased]: https://git.wilde-server.de/florian.wilde/netmover/compare/v0.5.0...HEAD
+[0.5.0]: https://git.wilde-server.de/florian.wilde/netmover/compare/v0.4.0...v0.5.0
 [0.4.0]: https://git.wilde-server.de/florian.wilde/netmover/compare/v0.3.0...v0.4.0
 [0.3.0]: https://git.wilde-server.de/florian.wilde/netmover/compare/v0.2.1...v0.3.0
 [0.2.1]: https://git.wilde-server.de/florian.wilde/netmover/compare/v0.2.0...v0.2.1
@@ -18,10 +18,37 @@ Zeiten in Ortszeit (Europe/Berlin).
 | 2026-10-10 | 23:31 | 23:58 | 0:27  | –       | Schlüsselverwaltung, Update-Skript, 0.2.x |
 | 2026-10-11 | 00:13 | 00:32 | 0:19  | –       | Menü, Anleitung, Roadmap, Uptime Kuma (0.3.0) |
 | 2026-10-11 | 00:31 | 00:50 | 0:19  | –       | Pausieren, Wartungsmodus, E-Mail (0.4.0) |
-| **Summe**  |       |       | **2:55** |      |                                        |
+| 2026-10-11 | 00:54 | 01:16 | 0:22  | –       | SFTP-Quelle, Zugangsdaten, Anmeldung (0.5.0) |
+| **Summe**  |       |       | **3:17** |      |                                        |

 ---

+## 2026-10-11 · 00:54–01:16 · SFTP-Quelle, Zugangsdaten, Anmeldung abschaltbar (0.5.0)
+
+**Erledigt**
+- `mover/sources.py`: `Source` mit `LocalSource`/`SftpSource`; Engine greift nur noch
+  darüber auf die Quelle zu (Scan, frischer Zustand, Lesen, Löschen, Quarantäne).
+  Ziele bekommen statt `Path` ein „Readable“ (Adapter) – Destinations fast unverändert.
+- `credentials.py` (vorher `keys.py`): gespeicherte Passwörter; Seite „Zugangsdaten“.
+- Anmeldung: Modus immer / vertraute Netze / nie, zentral im `guard`.
+- 26 neue Tests (325 gesamt); alle bisherigen Mover-Tests bestanden den Umbau
+  unverändert (nur ein Monkeypatch-Ziel verschoben).
+
+**Entscheidungen**
+- SFTP-Quelle: Stabilität über Größe + sekundengenaue mtime, Mindestfenster 2 s;
+  Verbindungsabbruch → `ENOTCONN` → Lauf abgebrochen (Anbindung gestört).
+- Gespeicherte Passwörter tragen den Benutzer selbst (anders als Schlüssel, deren
+  Benutzer zur Anbindung gehört).
+- „Passworteingabe deaktivieren“ als Login-Modus umgesetzt, mit Bestätigung durch
+  das Passwort, Warnhinweis und Notausgang über `reset-password`. Ein Passwort bleibt
+  Pflicht (Ersteinrichtung), sonst ließe sich die Anmeldung nie wieder einschalten.
+
+**Stolpersteine**
+- paramiko `prefetch()` braucht `fstat`; der Test-Server konnte es nicht – jetzt
+  kann er es, und die Quelle übergibt die bekannte Größe (spart einen Umlauf).
+- Zwei TestClients mit eigenem Lebenszyklus starteten den Scheduler doppelt →
+  zweiter Client ohne Lebenszyklus auf derselben App.
+
 ## 2026-10-11 · 00:31–00:50 · Pausieren, Wartungsmodus, E-Mail (0.4.0)

 **Erledigt**
@@ -3,7 +3,7 @@
 Selbst gehosteter Dienst mit Weboberfläche, der **SMB/CIFS- und NFS-Freigaben
 verwaltet** und **Dateien zeitgesteuert verschiebt** – „alle N Sekunden Ordner 1
 prüfen, passende Dateien nach Ordner 2 bringen“. Quelle und Ziel sind
-unabhängig voneinander per SMB, NFS, lokal oder (nur als Ziel) per SFTP angebunden.
+unabhängig voneinander per SMB, NFS, SFTP oder (als Ziel) lokal angebunden.

 Typischer Einsatz: Ein Synology-NAS nimmt Scans entgegen, NetMover schiebt sie alle
 30 Sekunden in den Eingangsordner von Paperless-ngx.
@@ -176,7 +176,7 @@ Lege in NetMover die *Anbindung → Neue Anbindung* an:
 | Feld | Wert |
 |---|---|
 | Name | `Paperless` |
-| Typ | SFTP (nur Ziel) |
+| Typ | SFTP |
 | Server / SSH-Port | IP des LXC / `22` |
 | Freigabe / Pfad | `/opt/paperless_data/consume` |
 | Anmeldeart | SSH-Schlüssel |
@@ -6,13 +6,6 @@ stehen mit Version im [CHANGELOG](CHANGELOG.md); Details zu Entscheidungen im

 Legende: ✅ erledigt · 🔨 in Arbeit · 📋 geplant · 💡 Idee (noch nicht entschieden)

-## In Arbeit (0.5.0)
-
-| Status | Thema | Warum |
-|---|---|---|
-| 🔨 | **SFTP als Quelle**: Scan, Stabilitätsprüfung und Löschen der Quelle über SFTP, mit denselben Garantien wie bei SMB/NFS | Quellen, die nur per SSH erreichbar sind |
-| 🔨 | **Gespeicherte Zugangsdaten** für SMB und SFTP-Passwörter, zusammen mit den SSH-Schlüsseln auf einer Seite „Zugangsdaten“ | Mehrere Freigaben mit demselben NAS-Benutzer an einer Stelle pflegen |
-
 ## Geplant

 | Status | Thema | Warum |
@@ -35,6 +28,7 @@ Legende: ✅ erledigt · 🔨 in Arbeit · 📋 geplant · 💡 Idee (noch nicht

 | Version | Thema |
 |---|---|
+| 0.5.0 | SFTP als Quelle; gespeicherte Zugangsdaten für SMB/SFTP-Passwörter (Seite „Zugangsdaten“); Anmeldung abschaltbar (vertrauenswürdige Netze / ganz) |
 | 0.4.0 | Automatisch pausieren/fortsetzen bei nicht erreichbaren Anbindungen; Wartungsmodus; E-Mail-Benachrichtigung; eine Meldung pro ausgefallener Anbindung |
 | 0.3.0 | Neues Menü (gruppierte Seitenleiste, Statusampel, Handy-Menü); Anleitung im Repo und als Hilfe-Seite; Roadmap; Uptime Kuma (Einzel-Endpunkte, Push-Monitor) |
 | 0.2.1 | Migrationen in einer Transaktion und ohne Fremdschlüsselfehler auf befüllten Datenbanken |
@@ -4,7 +4,7 @@
 services:
  netmover:
    build: .
-    image: netmover:0.4.0
+    image: netmover:0.5.0
    container_name: netmover
    restart: unless-stopped
    cap_add:
@@ -8,21 +8,21 @@ Diese Anleitung erklärt die Bedienung. Installation, Updates und technische Det
 stehen im [README](../README.md).

 **Inhalt:** [Begriffe](#begriffe) · [Erste Anmeldung](#erste-anmeldung) ·
-[Anbindungen](#anbindungen) · [Schlüssel](#schluessel) · [Regeln](#regeln) ·
+[Zugangsdaten](#zugangsdaten) · [Anbindungen](#anbindungen) · [Regeln](#regeln) ·
 [Probelauf](#probelauf) · [Status](#status) · [Wartungsmodus](#wartungsmodus) · [Protokoll](#protokoll) ·
-[Einstellungen](#einstellungen) · [Uptime Kuma](#uptime-kuma) · [Beispiele](#beispiele) · [Häufige Fragen](#fragen)
+[Einstellungen](#einstellungen) · [Anmeldung](#anmeldung) · [Uptime Kuma](#uptime-kuma) · [Beispiele](#beispiele) · [Häufige Fragen](#fragen)

 ## Die vier Begriffe {#begriffe}

 | Begriff | Was es ist | Beispiel |
 |---|---|---|
 | **Anbindung** | Ein Ort, an den NetMover herankommt: eine SMB-Freigabe, ein NFS-Export oder ein Server per SFTP | „NAS“ = `//10.0.10.6/Scans` |
-| **Schlüssel** | Ein gespeicherter SSH-Schlüssel für SFTP-Anbindungen | „paperless-lxc“ |
+| **Zugangsdaten** | Gespeicherter Benutzer mit Passwort oder SSH-Schlüssel, in mehreren Anbindungen nutzbar | „NAS-Scanner“, „paperless-lxc“ |
 | **Regel** | „Alle N Sekunden aus Anbindung A nach Anbindung B“ – mit Filter, Sicherheitsprüfungen und Verhalten bei gleichen Namen | „Scans → Paperless“ |
 | **Lauf** | Eine Ausführung einer Regel; steht mit allen Dateien im Protokoll | 23:41:07, 3 Dateien verschoben |

 Die Reihenfolge beim Einrichten ist immer dieselbe – und so ist auch das Menü
-aufgebaut: **Anbindungen → (Schlüssel) → Regeln → Status und Protokoll beobachten.**
+aufgebaut: **(Zugangsdaten) → Anbindungen → Regeln → Status und Protokoll beobachten.**

 ![Statusseite](img/status.png)

@@ -50,11 +50,13 @@ Passwort festlegt. Passwort ändern: *System → Einstellungen*.
 |---|---|---|
 | **SMB/CIFS** | Windows-Freigaben, Synology, QNAP, … | Quelle und Ziel |
 | **NFS 4 / NFS 3** | Linux-Server, NAS mit NFS | Quelle und Ziel |
-| **SFTP** | Server, die nur per SSH erreichbar sind (z. B. ein LXC mit Paperless) | nur Ziel |
+| **SFTP** | Server, die nur per SSH erreichbar sind (z. B. ein LXC mit Paperless) | Quelle und Ziel |

 SMB- und NFS-Anbindungen werden **eingehängt** und erscheinen unter
 `/mnt/netmover/<name>`. SFTP wird nicht eingehängt, NetMover verbindet sich bei
-jedem Lauf.
+jedem Lauf. Als **Quelle** gelten bei SFTP dieselben Sicherheitsprüfungen; weil
+SFTP-Server Änderungszeiten nur sekundengenau melden, sollte das
+Stabilitätsfenster dort mindestens 2 Sekunden betragen.

 ### Felder {#anbindung-felder}

@@ -64,7 +66,7 @@ jedem Lauf.
 | **Server** | IP-Adresse oder Hostname, z. B. `10.0.10.6` |
 | **Freigabe / Pfad** | SMB: Freigabename, optional mit Unterordner (`Scans` oder `Scans/privat`). NFS: Exportpfad (`/volume1/scans`). SFTP: absoluter Ordner auf dem Server (`/opt/paperless_data/consume`). |
 | **Benutzername / Passwort** | Bei SMB leer lassen für Gastzugang. Beim Bearbeiten heißt ein leeres Passwortfeld „unverändert“ – Passwörter werden nie angezeigt. |
-| **Anmeldeart, Schlüssel** (SFTP) | SSH-Schlüssel (gespeichert oder nur für diese Anbindung) oder Passwort, siehe [Schlüssel](#schluessel). |
+| **Zugangsdaten / Anmeldeart** | Gespeicherte Zugangsdaten auswählen oder eigene Angaben nur für diese Anbindung; bei SFTP zusätzlich SSH-Schlüssel oder Passwort, siehe [Zugangsdaten](#zugangsdaten). |
 | **uid, gid, file_mode, dir_mode** (SMB) | Wem die Dateien im eingehängten Ordner gehören. Meist leer lassen. |
 | **Zusatzoptionen** | Für Sonderfälle, z. B. `vers=3.0` oder `nohandlecache`. Nur erlaubte Optionen; Zugangsdaten setzt NetMover selbst. |
 | **Beim Start automatisch einhängen** | Nach einem Neustart des Servers wieder einhängen. Fällt das NAS aus, versucht NetMover es außerdem jede Minute erneut. |
@@ -102,14 +104,28 @@ Es wird nie in ein leeres Verzeichnis geschrieben, nur weil die Freigabe fehlt.
 Solche Regeln **warten** automatisch, bis die Anbindung zurück ist (siehe
 [Status](#status)), und laufen dann von selbst weiter.

-## Schlüssel {#schluessel}
+## Zugangsdaten {#zugangsdaten}

-*Verbindungen → Schlüssel*
+*Verbindungen → Zugangsdaten*

-Für SFTP-Ziele ist ein SSH-Schlüssel sicherer als ein Passwort. Einen Schlüssel kann
-man für mehrere Anbindungen verwenden; den Benutzernamen legt jede Anbindung fest.
+Benutzer und Passwörter für Freigaben sowie SSH-Schlüssel einmal hinterlegen und in
+beliebig vielen Anbindungen auswählen. Passwörter und private Schlüssel liegen
+verschlüsselt in der Datenbank und werden nie angezeigt.

-![Schlüssel](img/schluessel.png)
+![Zugangsdaten](img/zugangsdaten.png)
+
+### Passwörter {#zugangsdaten-passwoerter}
+
+Für SMB-Freigaben und SFTP-Server mit Passwort: Name, Benutzer, Passwort und
+optional Domäne. In der Anbindung unter *Zugangsdaten* auswählen statt „eigene
+Angaben“. **Ändern** setzt Benutzer, Domäne oder ein neues Passwort – das gilt sofort
+für alle Anbindungen, die den Eintrag nutzen (praktisch, wenn das NAS-Passwort
+wechselt).
+
+### SSH-Schlüssel {#schluessel}
+
+Für SFTP ist ein Schlüssel sicherer als ein Passwort. Einen Schlüssel kann man für
+mehrere Anbindungen verwenden; den Benutzernamen legt jede Anbindung fest.

 **Neuen Schlüssel erzeugen** – Namen eingeben, *Ed25519-Schlüssel erzeugen*. Dann:

@@ -117,17 +133,15 @@ man für mehrere Anbindungen verwenden; den Benutzernamen legt jede Anbindung fe
   **authorized_keys-Zeile** kopieren. Sie beginnt mit
   `restrict,command="/usr/lib/openssh/sftp-server"` – damit darf der Schlüssel
   **nur SFTP**, keine Shell.
-2. Auf dem Zielserver an die Datei `~/.ssh/authorized_keys` des Benutzers anhängen
+2. Auf dem Server an die Datei `~/.ssh/authorized_keys` des Benutzers anhängen
   (Ordner `~/.ssh` mit Rechten `700`, Datei `600`, Eigentümer der Benutzer).
 3. In der Anbindung *Anmeldeart: SSH-Schlüssel* und den Schlüssel wählen, **Testen**.

 **Vorhandenen Schlüssel importieren** – privaten Schlüssel einfügen (OpenSSH- oder
 PEM-Format; Ed25519, ECDSA oder RSA ab 2048 Bit). Hat er eine Passphrase, diese
-angeben: Sie wird nur zum Entschlüsseln benutzt, der Schlüssel liegt danach
-verschlüsselt in NetMovers Datenbank.
+angeben: Sie wird nur zum Entschlüsseln benutzt.

-Private Schlüssel zeigt NetMover nie an. Ein Schlüssel lässt sich nur löschen, wenn
-ihn keine Anbindung mehr nutzt.
+Gespeicherte Zugangsdaten lassen sich nur löschen, wenn sie keine Anbindung mehr nutzt.

 ## Regeln {#regeln}

@@ -317,6 +331,28 @@ Als Problem gilt: Scheduler steht, eine aktive Anbindung ist im Fehlerzustand, e
 aktive Regel hatte länger als das Dreifache ihres Intervalls keinen Erfolg oder ist
 dreimal hintereinander gescheitert.

+### Anmeldung {#anmeldung}
+
+Standardmäßig fragt NetMover immer nach dem Passwort. Unter *Einstellungen →
+Anmeldung* lässt sich das ändern (Bestätigung mit dem aktuellen Passwort):
+
+| Einstellung | Wirkung |
+|---|---|
+| **immer** (empfohlen) | Anmeldung bei jedem Zugriff |
+| **nicht aus vertrauenswürdigen Netzen** | ohne Passwort aus den eingetragenen Netzen, z. B. `10.0.10.0/24`; von überall sonst mit Passwort |
+| **nie** | keine Anmeldung – nur hinter einem Reverse Proxy mit eigener Anmeldung (z. B. Authelia, Authentik) |
+
+Solange ohne Anmeldung gearbeitet wird, steht oben gelb „ohne Anmeldung“. Der Schutz
+gegen untergeschobene Formulare bleibt immer aktiv.
+
+> NetMover läuft als root und kennt die Zugangsdaten deiner Freigaben. Wer die
+> Oberfläche ohne Anmeldung erreicht, kann alles ändern. Läuft ein Reverse Proxy auf
+> demselben Server, kommen alle Anfragen von `127.0.0.1` – dann `127.0.0.1` nicht als
+> vertrauenswürdig eintragen und `NETMOVER_FORWARDED_ALLOW_IPS` setzen.
+
+Zurück zur Anmeldung: dieselbe Einstellung auf „immer“, oder auf dem Server
+`/opt/netmover/venv/bin/netmover reset-password` (setzt auch die Anmeldung zurück).
+
 ### Export und Import {#export-import}

 *Konfiguration exportieren* lädt Anbindungen, Regeln und Einstellungen als JSON
@@ -329,8 +365,8 @@ deaktiviert, Zugangsdaten müssen neu eingegeben werden.
 ### Scanner → NAS → Paperless {#beispiel-paperless}

 1. Anbindung **NAS** (SMB, Freigabe `Scans`), testen, einhängen.
-2. Schlüssel **paperless-lxc** erzeugen, Zeile in `authorized_keys` im Paperless-LXC
-   eintragen.
+2. Unter *Zugangsdaten* den Schlüssel **paperless-lxc** erzeugen, Zeile in
+   `authorized_keys` im Paperless-LXC eintragen.
 3. Anbindung **Paperless** (SFTP, Pfad `/opt/paperless_data/consume`, Schlüssel
   wählen, Benutzer = der Paperless-Benutzer), testen.
 4. Regel **Scans → Paperless**: Quelle NAS, Filter `*.pdf`, Ziel Paperless,
@@ -1,3 +1,3 @@
 """NetMover – NFS/SMB-Mounts verwalten und Dateien zeitgesteuert verschieben."""

-__version__ = "0.4.0"
+__version__ = "0.5.0"
@@ -85,6 +85,7 @@ def _int(


 NEW_KEY = "new"
+OWN = "own"


@dataclass
@@ -101,6 +102,8 @@ class MountInput:
    # nur sftp + Schlüssel: "new" = eigener Schlüssel nur für diese Anbindung,
    # sonst ID eines Schlüssels aus der Schlüsselverwaltung
    key_id: str = NEW_KEY
+    # Passwort-Anmeldung (SMB, SFTP): "own" = eigene Angaben, sonst ID gespeicherter Zugangsdaten
+    cred_id: str = OWN
    uid: int | None = None
    gid: int | None = None
    file_mode: str | None = None
@@ -135,7 +138,12 @@ def mount_input_from_model(m: Mount) -> MountInput:
        username=m.username or (c.username if c else ""),
        domain=(c.domain or "") if c else "",
        auth=c.kind if c else CredentialKind.PASSWORD,
-        key_id=str(c.id) if c is not None and c.managed else NEW_KEY,
+        key_id=str(c.id)
+        if c is not None and c.managed and c.kind is CredentialKind.SSH_KEY
+        else NEW_KEY,
+        cred_id=str(c.id)
+        if c is not None and c.managed and c.kind is CredentialKind.PASSWORD
+        else OWN,
        uid=m.uid,
        gid=m.gid,
        file_mode=m.file_mode,
@@ -159,10 +167,14 @@ def parse_mount_form(form: Mapping[str, str]) -> tuple[MountInput, Errors]:
        auto_mount=_bool(form, "auto_mount"),
        enabled=_bool(form, "enabled"),
        key_id=(form.get("key_id") or NEW_KEY).strip(),
+        cred_id=(form.get("cred_id") or OWN).strip(),
    )
    if data.key_id != NEW_KEY and not data.key_id.isdigit():
        errors["key_id"] = "Ungültige Schlüsselauswahl."
        data.key_id = NEW_KEY
+    if data.cred_id != OWN and not data.cred_id.isdigit():
+        errors["cred_id"] = "Ungültige Auswahl."
+        data.cred_id = OWN
    try:
        data.type = MountType(form.get("type", "cifs"))
    except ValueError:
@@ -210,7 +222,14 @@ def parse_mount_form(form: Mapping[str, str]) -> tuple[MountInput, Errors]:
        data.username = check("username", lambda: validate_username(data.username), data.username)
    if data.domain:
        check("domain", lambda: validate_username(data.domain), "")
-    if data.type is MountType.SFTP and not data.username:
+    if data.auth is CredentialKind.SSH_KEY or data.type in (MountType.NFS3, MountType.NFS4):
+        data.cred_id = OWN
+    uses_stored = data.cred_id != OWN
+    if uses_stored:
+        data.username = data.domain = data.password = ""  # kommen aus den Zugangsdaten
+        errors.pop("username", None)
+        errors.pop("password", None)
+    if data.type is MountType.SFTP and not data.username and not uses_stored:
        errors["username"] = "SFTP braucht einen Benutzernamen."
    if data.type is not MountType.SFTP:
        data.auth = CredentialKind.PASSWORD
@@ -224,7 +243,19 @@ def credential_for_test(
 ) -> CredentialData | None:
    """Zugangsdaten für „Testen“ aus dem Formular – leeres Passwort = gespeichertes,
    gewählter Schlüssel aus der Schlüsselverwaltung."""
-    if data.type in (MountType.NFS3, MountType.NFS4) or not data.username:
+    if data.type in (MountType.NFS3, MountType.NFS4):
+        return None
+    if data.cred_id != OWN and s is not None:
+        stored = s.get(Credential, int(data.cred_id))
+        if stored is None or not stored.managed or stored.kind is not CredentialKind.PASSWORD:
+            return None
+        return CredentialData(
+            CredentialKind.PASSWORD,
+            stored.username,
+            secrets.decrypt(stored.secret_enc),
+            stored.domain,
+        )
+    if not data.username:
        return None
    if data.auth is CredentialKind.SSH_KEY and data.key_id != NEW_KEY and s is not None:
        key = s.get(Credential, int(data.key_id))
@@ -321,8 +352,12 @@ def save_mount(
    # (geteilt, wird hier nie verändert) oder eigene Zugangsdaten nur dieser Anbindung.
    current = m.credential
    own = current if current is not None and not current.managed else None
+    stored_password = data.cred_id != OWN and (
+        data.type is MountType.CIFS
+        or (data.type is MountType.SFTP and data.auth is CredentialKind.PASSWORD)
+    )
    needs_cred = data.type is MountType.SFTP or (
-        data.type is MountType.CIFS and bool(data.username)
+        data.type is MountType.CIFS and (bool(data.username) or stored_password)
    )
    use_managed = (
        data.type is MountType.SFTP
@@ -332,6 +367,13 @@ def save_mount(
    if not needs_cred:
        m.credential = None
        m.username = None
+    elif stored_password:
+        stored = s.get(Credential, int(data.cred_id))
+        if stored is None or not stored.managed or stored.kind is not CredentialKind.PASSWORD:
+            s.rollback()
+            return None, {"cred_id": "Zugangsdaten existieren nicht."}
+        m.credential = stored
+        m.username = None
    elif use_managed:
        key = s.get(Credential, int(data.key_id))
        if key is None or not key.managed or key.kind is not CredentialKind.SSH_KEY:
@@ -496,8 +538,6 @@ def parse_rule_form(s: Session, form: Mapping[str, str]) -> tuple[RuleInput, Err
    src = s.get(Mount, d.src_mount_id) if d.src_mount_id else None
    if src is None:
        errors["src_mount_id"] = "Quelle wählen."
-    elif not src.type.is_kernel_mount:
-        errors["src_mount_id"] = "SFTP-Anbindungen sind nur als Ziel möglich."
    try:
        d.src_subpath = safe_relpath("src_subpath", d.src_subpath)
    except ValidationError as exc:
@@ -530,6 +570,10 @@ def parse_rule_form(s: Session, form: Mapping[str, str]) -> tuple[RuleInput, Err
            errors["pattern"] = f"Ungültiger regulärer Ausdruck: {exc}"
    if d.stability_seconds >= d.op_timeout_seconds:
        errors["stability_seconds"] = "Muss kleiner als das Zeitlimit sein."
+    elif src is not None and src.type is MountType.SFTP and d.stability_seconds < 2:
+        errors["stability_seconds"] = (
+            "Bei SFTP-Quellen mindestens 2 s – der Server meldet Zeiten nur sekundengenau."
+        )
    if d.set_owner and not _OWNER_RE.fullmatch(d.set_owner):
        errors["set_owner"] = "Als „uid:gid“ angeben, z. B. „1000:1000“."
    try:
@@ -64,6 +64,10 @@ def main(argv: Sequence[str] | None = None) -> int:
            row = s.get(Setting, PASSWORD_KEY)
            if row is not None:
                s.delete(row)
+            # Anmeldung wieder einschalten (falls abgeschaltet)
+            mode = s.get(Setting, "app.auth_mode")
+            if mode is not None:
+                mode.value = '"always"'
            s.commit()
        engine.dispose()
        print("Passwort gelöscht. Dienst neu starten; der Einrichtungscode steht im Log.")
@@ -1,4 +1,9 @@
-"""Schlüsselverwaltung: gespeicherte SSH-Schlüssel für SFTP-Anbindungen.
+"""Zugangsdaten-Verwaltung: gespeicherte Passwörter und SSH-Schlüssel.
+
+Passwörter (SMB und SFTP): Benutzer, Passwort, optional Domäne – einmal gepflegt,
+von beliebig vielen Anbindungen genutzt. Ein neues Passwort gilt sofort für alle.
+
+SSH-Schlüssel:

 * Erzeugen (Ed25519) oder Importieren (Ed25519, ECDSA, RSA ≥ 2048 Bit; OpenSSH-
  oder PEM-Format, optional mit Passphrase). Eine Passphrase wird beim Import
@@ -63,7 +68,7 @@ def _validate_name(s: Session, name: str, exclude_id: int | None = None) -> str:
        raise AdminError("Namen mit „mount:“ sind für NetMover reserviert.")
    dup = s.scalar(select(Credential.id).where(Credential.name == name))
    if dup is not None and dup != exclude_id:
-        raise AdminError(f"Ein Schlüssel „{name}“ existiert bereits.")
+        raise AdminError(f"Zugangsdaten mit dem Namen „{name}“ existieren bereits.")
    return name


@@ -153,9 +158,115 @@ def _users(s: Session, key_id: int) -> list[str]:
    return sorted(s.scalars(select(Mount.name).where(Mount.cred_id == key_id)))


+# -- Passwörter -------------------------------------------------------------------------------
+
+
+@dataclass
+class PasswordInfo:
+    id: int
+    name: str
+    username: str
+    domain: str | None
+    created_at: datetime
+    used_by: list[str] = field(default_factory=list)
+
+
+def _validate_login(username: str, password: str | None, domain: str) -> None:
+    from netmover.validation import ValidationError, validate_secret, validate_username
+
+    try:
+        validate_username(username)
+        if domain:
+            validate_username(domain)
+        if password is not None:
+            validate_secret(password)
+    except ValidationError as exc:
+        raise AdminError(str(exc)) from None
+
+
+def create_password(
+    s: Session, secrets: SecretBox, name: str, username: str, password: str, domain: str = ""
+) -> Credential:
+    name = _validate_name(s, name)
+    username, domain = username.strip(), domain.strip()
+    if not password:
+        raise AdminError("Passwort fehlt.")
+    _validate_login(username, password, domain)
+    cred = Credential(
+        name=name,
+        kind=CredentialKind.PASSWORD,
+        username=username,
+        domain=domain or None,
+        secret_enc=secrets.encrypt(password),
+        managed=True,
+    )
+    s.add(cred)
+    s.flush()
+    return cred
+
+
+def update_password(
+    s: Session, secrets: SecretBox, cred_id: int, username: str, password: str, domain: str = ""
+) -> Credential:
+    """Benutzer/Domäne ändern; leeres Passwort = unverändert."""
+    cred = managed_password(s, cred_id)
+    username, domain = username.strip(), domain.strip()
+    _validate_login(username, password or None, domain)
+    cred.username = username
+    cred.domain = domain or None
+    if password:
+        cred.secret_enc = secrets.encrypt(password)
+    return cred
+
+
+def managed_password(s: Session, cred_id: int) -> Credential:
+    cred = s.get(Credential, cred_id)
+    if cred is None or not cred.managed or cred.kind is not CredentialKind.PASSWORD:
+        raise AdminError("Zugangsdaten existieren nicht.")
+    return cred
+
+
+def list_passwords(s: Session) -> list[PasswordInfo]:
+    rows = s.scalars(
+        select(Credential)
+        .where(Credential.managed.is_(True), Credential.kind == CredentialKind.PASSWORD)
+        .order_by(Credential.name)
+    ).all()
+    return [
+        PasswordInfo(c.id, c.name, c.username, c.domain, c.created_at, _users(s, c.id))
+        for c in rows
+    ]
+
+
+def delete_credential(s: Session, cred_id: int) -> str:
+    """Gespeicherte Zugangsdaten (Passwort oder Schlüssel) löschen – nur unbenutzt."""
+    cred = s.get(Credential, cred_id)
+    if cred is None or not cred.managed:
+        raise AdminError("Zugangsdaten existieren nicht.")
+    users = _users(s, cred_id)
+    if users:
+        raise AdminError(f"„{cred.name}“ wird noch benutzt von: {', '.join(users)}")
+    name = cred.name
+    s.delete(cred)
+    s.flush()
+    return name
+
+
+def rename_credential(s: Session, cred_id: int, name: str) -> None:
+    cred = s.get(Credential, cred_id)
+    if cred is None or not cred.managed:
+        raise AdminError("Zugangsdaten existieren nicht.")
+    cred.name = _validate_name(s, name, exclude_id=cred_id)
+
+
+# -- SSH-Schlüssel ------------------------------------------------------------------------------
+
+
 def list_keys(s: Session) -> list[KeyInfo]:
    keys = s.scalars(
-        select(Credential).where(Credential.managed.is_(True)).order_by(Credential.name)
+        select(Credential)
+        .where(Credential.managed.is_(True), Credential.kind == CredentialKind.SSH_KEY)
+        .order_by(Credential.name)
    ).all()
    result = []
    for k in keys:
@@ -30,12 +30,13 @@ from abc import ABC, abstractmethod
 from collections.abc import Callable
 from dataclasses import dataclass, field
 from pathlib import Path
-from typing import Literal
+from typing import Any, Literal, Protocol

 from netmover.models import CollisionPolicy
 from netmover.mounting.sftp import SftpSession
 from netmover.mover.fsops import rename_noreplace
 from netmover.mover.naming import collision_candidates
+from netmover.mover.sources import network_errors
 from netmover.timeouts import CancelToken

 CHUNK = 1024 * 1024
@@ -81,6 +82,13 @@ class CommitResult:
 Verify = Callable[[], None]


+class Readable(Protocol):
+    """Quelle einer Übertragung: ein lokaler ``Path`` oder eine Datei einer
+    :class:`~netmover.mover.sources.Source` (über einen kleinen Adapter)."""
+
+    def open(self, mode: str = ...) -> Any: ...
+
+
 class Destination(ABC):
    kind: FsKind

@@ -107,7 +115,7 @@ class Destination(ABC):
    def _ensure_parent(self, rel: str) -> None: ...

    @abstractmethod
-    def _upload(self, src: Path, tmp: str, size: int, token: CancelToken) -> list[str]: ...
+    def _upload(self, src: Readable, tmp: str, size: int, token: CancelToken) -> list[str]: ...

    @abstractmethod
    def _apply_meta(self, rel: str, meta: MetaOptions) -> list[str]: ...
@@ -140,7 +148,7 @@ class Destination(ABC):

    def commit(
        self,
-        src: Path,
+        src: Readable,
        size: int,
        rel: str,
        policy: CollisionPolicy,
@@ -234,7 +242,7 @@ class LocalDestination(Destination):
    def _ensure_parent(self, rel: str) -> None:
        self._p(rel).parent.mkdir(mode=0o755, parents=True, exist_ok=True)

-    def _upload(self, src: Path, tmp: str, size: int, token: CancelToken) -> list[str]:
+    def _upload(self, src: Readable, tmp: str, size: int, token: CancelToken) -> list[str]:
        written = 0
        fd = os.open(self._p(tmp), os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
        try:
@@ -429,10 +437,10 @@ class SftpDestination(Destination):
            except FileNotFoundError:
                self.client.mkdir(current, mode=0o755)

-    def _upload(self, src: Path, tmp: str, size: int, token: CancelToken) -> list[str]:
+    def _upload(self, src: Readable, tmp: str, size: int, token: CancelToken) -> list[str]:
        notes: list[str] = []
        written = 0
-        with self.client.open(self._p(tmp), "wb") as fh, src.open("rb") as fin:
+        with network_errors(), self.client.open(self._p(tmp), "wb") as fh, src.open("rb") as fin:
            fh.set_pipelined(True)
            while chunk := fin.read(CHUNK // 4):
                token.check()
@@ -15,12 +15,12 @@ from __future__ import annotations

 import dataclasses
 import errno
-import os
-import stat
+import posixpath
 import time
 from collections.abc import Callable
 from dataclasses import dataclass, field
 from pathlib import Path
+from typing import Any

 from netmover.logging_setup import event, get_logger
 from netmover.models import (
@@ -40,7 +40,8 @@ from netmover.mover.destinations import (
 )
 from netmover.mover.ledger import FailureInfo, Ledger
 from netmover.mover.naming import collision_candidates
-from netmover.mover.scan import BUSY, Candidate, FileState, Matcher, current_state, scan
+from netmover.mover.scan import BUSY, Candidate, FileState, Matcher
+from netmover.mover.sources import LocalSource, Source, SourceMissing
 from netmover.timeouts import Cancelled, CancelToken

 log = get_logger(__name__)
@@ -114,6 +115,17 @@ def _mb(n: int) -> str:
    return f"{n / 1024 / 1024:.1f} MB"


+@dataclass(frozen=True)
+class _SourceFile:
+    """Adapter: eine Datei einer :class:`Source` mit ``open()`` wie ein ``Path``."""
+
+    source: Source
+    rel: str
+
+    def open(self, mode: str = "rb") -> Any:
+        return self.source.open(self.rel)
+
+
 class Engine:
    def __init__(
        self,
@@ -140,20 +152,27 @@ class Engine:
    def execute(
        self,
        rule: RuleSnapshot,
-        src_root: Path,
+        source: Source | Path,
        dest: Destination,
        token: CancelToken,
        *,
        dry_run: bool = False,
+        quarantine_rel: str | None = None,
        quarantine_root: Path | None = None,
    ) -> RunResult:
+        """``source``: eine :class:`Source` oder (Kurzform) ein lokales Verzeichnis.
+        Quarantäne relativ zur Wurzel der Anbindung (``quarantine_rel``) oder – bei
+        der Kurzform – als Pfad innerhalb des Verzeichnisses (``quarantine_root``)."""
        result = RunResult(dry_run=dry_run)
+        src = source if isinstance(source, Source) else LocalSource(source)
+        if quarantine_root is not None and isinstance(src, LocalSource):
+            quarantine_rel = str(quarantine_root.relative_to(src.root))
        try:
-            self._execute(rule, src_root, dest, token, result, dry_run, quarantine_root)
+            self._execute(rule, src, dest, token, result, dry_run, quarantine_rel)
        except Cancelled:
            result.status = RunStatus.ERROR
            result.message = "Lauf abgebrochen"
-        except RunAborted as exc:
+        except (RunAborted, SourceMissing) as exc:
            result.status = RunStatus.ERROR
            result.message = str(exc)
        except TransferError as exc:
@@ -170,26 +189,23 @@ class Engine:
    def _execute(
        self,
        rule: RuleSnapshot,
-        src_root: Path,
+        src: Source,
        dest: Destination,
        token: CancelToken,
        result: RunResult,
        dry_run: bool,
-        quarantine_root: Path | None,
+        quarantine_rel: str | None,
    ) -> None:
-        if not src_root.is_dir():
-            raise RunAborted(f"Quellordner {src_root} existiert nicht")
+        src.check_root()
        if not dry_run:
            dest.prepare()
        token.check()

-        exclude: set[Path] = set()
-        if quarantine_root is not None:
-            exclude.add(quarantine_root.resolve())
-        if isinstance(dest, LocalDestination):
-            exclude.add(dest.root.resolve())  # Ziel innerhalb der Quelle nie erneut scannen
+        exclude = frozenset({quarantine_rel}) if quarantine_rel else frozenset()
+        if isinstance(dest, LocalDestination) and isinstance(src, LocalSource):
+            src.exclude_paths.add(dest.root.resolve())  # Ziel innerhalb der Quelle auslassen
        matcher = Matcher(rule.pattern, rule.pattern_type)
-        candidates = list(scan(src_root, matcher, rule.recursive, frozenset(exclude)))
+        candidates = src.scan(matcher, rule.recursive, exclude)
        result.seen = len(candidates)
        token.check()

@@ -227,13 +243,13 @@ class Engine:

        # Stabilität (Fallstrick 1): zwei frische Messungen (öffnen + fstat, nicht
        # der Verzeichnis-Cache) im Abstand des Stabilitätsfensters.
-        first: dict[str, FileState | None] = {c.relpath: current_state(c.path) for c in pending}
+        first: dict[str, FileState | None] = {c.relpath: src.state(c.relpath) for c in pending}
        if pending and rule.stability_seconds > 0:
            self._sleep(rule.stability_seconds)
            token.check()
        stable: list[Candidate] = []
        for c in pending:
-            before, st = first[c.relpath], current_state(c.path)
+            before, st = first[c.relpath], src.state(c.relpath)
            if before is None or st is None:
                continue  # inzwischen verschwunden – nichts zu tun
            if st != before or st == BUSY:
@@ -255,7 +271,7 @@ class Engine:
        failures = self.ledger.failures(rule.id)
        for c in stable:
            token.check()
-            result.add(self._process(rule, c, dest, token, dry_run, quarantine_root, failures))
+            result.add(self._process(rule, c, src, dest, token, dry_run, quarantine_rel, failures))

        if unchanged:
            result.message = f"{unchanged} bereits übertragen"
@@ -263,9 +279,9 @@ class Engine:
    # -- einzelne Datei ------------------------------------------------------------------

    def _meta(
-        self, rule: RuleSnapshot, path: Path, dest: Destination
+        self, rule: RuleSnapshot, src: Source, rel: str, dest: Destination
    ) -> tuple[MetaOptions, list[str]]:
-        st = os.stat(path)
+        st = src.stat(rel)
        notes: list[str] = []
        owner = rule.set_owner
        if owner is not None:
@@ -280,17 +296,18 @@ class Engine:
                if self._warn_once(f"{rule.id}:nfs-chown"):
                    reason = "Root-Squash aktiv" if rule.dst_root_squash else "Schreibtest fehlt"
                    notes.append(f"NFS-Ziel: chown übersprungen ({reason})")
-        mode = rule.set_mode if rule.set_mode is not None else stat.S_IMODE(st.st_mode)
-        return MetaOptions(st.st_atime_ns, st.st_mtime_ns, mode, owner), notes
+        mode = rule.set_mode if rule.set_mode is not None else st.mode
+        return MetaOptions(st.atime_ns, st.mtime_ns, mode, owner), notes

    def _process(
        self,
        rule: RuleSnapshot,
        c: Candidate,
+        src: Source,
        dest: Destination,
        token: CancelToken,
        dry_run: bool,
-        quarantine_root: Path | None,
+        quarantine_rel: str | None,
        failures: dict[str, FailureInfo],
    ) -> ItemResult:
        size = c.state.size
@@ -298,7 +315,7 @@ class Engine:

        # Quarantäne für wiederholt scheiternde Dateien
        if (
-            quarantine_root is not None
+            quarantine_rel is not None
            and failure is not None
            and failure.attempts >= rule.quarantine_after
            and (failure.size, failure.mtime_ns) == (c.state.size, c.state.mtime_ns)
@@ -307,11 +324,11 @@ class Engine:
                return ItemResult(
                    c.relpath,
                    ItemStatus.PLANNED,
-                    str(quarantine_root / c.relpath),
+                    src.describe_root_rel(posixpath.join(quarantine_rel, c.relpath)),
                    size,
                    "würde in Quarantäne verschoben",
                )
-            return self._quarantine(rule, c, quarantine_root, failure.attempts)
+            return self._quarantine(rule, c, src, quarantine_rel, failure.attempts)

        try:
            planned = dest.plan(c.relpath, rule.on_collision)
@@ -351,16 +368,25 @@ class Engine:
            )

        def verify() -> None:
-            if current_state(c.path) != c.state:
+            if src.state(c.relpath) != c.state:
                raise SourceChanged("Quelle hat sich während der Übertragung geändert")

        try:
-            meta, notes = self._meta(rule, c.path, dest)
+            meta, notes = self._meta(rule, src, c.relpath, dest)
            res = None
-            if rule.action is RuleAction.MOVE:
-                res = dest.move_same_device(c.path, c.relpath, rule.on_collision, meta)
+            local = src.local_path(c.relpath)
+            if rule.action is RuleAction.MOVE and local is not None:
+                res = dest.move_same_device(local, c.relpath, rule.on_collision, meta)
            if res is None:
-                res = dest.commit(c.path, size, c.relpath, rule.on_collision, meta, verify, token)
+                res = dest.commit(
+                    _SourceFile(src, c.relpath),
+                    size,
+                    c.relpath,
+                    rule.on_collision,
+                    meta,
+                    verify,
+                    token,
+                )
        except CollisionSkip:
            return ItemResult(
                c.relpath,
@@ -385,7 +411,7 @@ class Engine:

        if not res.source_consumed:
            try:
-                os.unlink(c.path)
+                src.delete(c.relpath)
            except OSError as exc:
                # Ziel ist vollständig – Buchführung verhindert Duplikate im nächsten Lauf.
                self.ledger.record_copy(rule.id, c.relpath, size, c.state.mtime_ns, target)
@@ -412,15 +438,18 @@ class Engine:
        return ItemResult(c.relpath, ItemStatus.FAILED, None, c.state.size, msg)

    def _quarantine(
-        self, rule: RuleSnapshot, c: Candidate, root: Path, attempts: int
+        self, rule: RuleSnapshot, c: Candidate, src: Source, q_rel: str, attempts: int
    ) -> ItemResult:
+        target = ""
        try:
            for cand in collision_candidates(c.relpath):
-                target = root / cand
-                if os.path.lexists(target):
+                target = posixpath.join(q_rel, cand)
+                if src.exists_root_rel(target):
+                    continue
+                try:
+                    src.move_to(c.relpath, target)
+                except FileExistsError:
                    continue
-                target.parent.mkdir(mode=0o755, parents=True, exist_ok=True)
-                os.rename(c.path, target)
                break
            else:
                raise TransferError("kein freier Name im Quarantäneordner")
@@ -434,7 +463,7 @@ class Engine:
        return ItemResult(
            c.relpath,
            ItemStatus.QUARANTINED,
-            str(target),
+            src.describe_root_rel(target),
            c.state.size,
            f"nach {attempts} Fehlversuchen in Quarantäne verschoben",
        )
@@ -34,6 +34,7 @@ from netmover.mover.destinations import (
 )
 from netmover.mover.engine import Engine, RuleSnapshot, RunResult
 from netmover.mover.ledger import Ledger
+from netmover.mover.sources import LocalSource, SftpSource, Source
 from netmover.timeouts import CancelToken, OperationTimeout, run_with_watchdog
 from netmover.validation import ValidationError, safe_abspath, safe_relpath

@@ -46,6 +47,7 @@ class RulePlan:
    name: str
    src_mount_id: int
    src_type: MountType
+    src_name: str
    src_subpath: str
    dst_mount_id: int | None
    dst_type: MountType | None
@@ -96,6 +98,7 @@ def load_plan(session: Session, rule_id: int) -> RulePlan:
        name=r.name,
        src_mount_id=r.src_mount_id,
        src_type=r.src_mount.type,
+        src_name=r.src_mount.name,
        src_subpath=safe_relpath("src_subpath", r.src_subpath),
        dst_mount_id=r.dst_mount_id,
        dst_type=dst.type if dst is not None else None,
@@ -166,10 +169,6 @@ class RuleRunner:
        rule_id: int,
        handed_over: list[bool],
    ) -> RunResult:
-        if not plan.src_type.is_kernel_mount:
-            return RunResult(
-                RunStatus.ERROR, dry, message="SFTP-Anbindungen sind nur als Ziel möglich"
-            )
        try:
            src_mp = self.mounts.check_alive(plan.src_mount_id)
            dst_mp = (
@@ -179,29 +178,33 @@ class RuleRunner:
            )
        except MountUnavailable as exc:
            return RunResult(RunStatus.ERROR, dry, message=str(exc), unavailable=True)
-        if src_mp is None:  # nur bei SFTP, oben bereits ausgeschlossen
-            return RunResult(RunStatus.ERROR, dry, message="Quelle ist kein Kernel-Mount")
-        src_root = src_mp / plan.src_subpath if plan.src_subpath else src_mp
-        quarantine_root = src_mp / plan.quarantine_path if plan.quarantine_path else None

        try:
+            make_source = self._source_factory(plan, src_mp)
+            src_root = src_mp / plan.src_subpath if src_mp and plan.src_subpath else src_mp
            make_dest = self._dest_factory(plan, dst_mp, src_root)
        except (ValidationError, TransferError, SftpError) as exc:
            return RunResult(RunStatus.ERROR, dry, message=str(exc))

        def work(token: CancelToken) -> RunResult:
+            source = make_source()
+            try:
                dest = make_dest()
+            except BaseException:
+                source.close()
+                raise
            try:
                return self.engine.execute(
                    plan.snapshot,
-                    src_root,
+                    source,
                    dest,
                    token,
                    dry_run=dry,
-                    quarantine_root=quarantine_root,
+                    quarantine_rel=plan.quarantine_path,
                )
            finally:
                dest.close()
+                source.close()

        def finish() -> None:
            self._started.pop(rule_id, None)
@@ -230,8 +233,26 @@ class RuleRunner:
        paths = {path.absolute(), path.resolve()}
        return any(p == b or b in p.parents for p in paths for b in bases)

+    def _source_factory(self, plan: RulePlan, src_mp: Path | None) -> Callable[[], Source]:
+        if plan.src_type is MountType.SFTP:
+            scfg = self.mounts.sftp_config_for(plan.src_mount_id)
+            label = plan.src_name or "sftp"
+
+            def make_sftp() -> Source:
+                return SftpSource(connect(scfg), scfg.base_path, plan.src_subpath, label=label)
+
+            return make_sftp
+        if src_mp is None:
+            raise TransferError("Quell-Anbindung ist nicht eingehängt")
+        root = src_mp
+
+        def make_local() -> Source:
+            return LocalSource(root, plan.src_subpath)
+
+        return make_local
+
    def _dest_factory(
-        self, plan: RulePlan, dst_mp: Path | None, src_root: Path
+        self, plan: RulePlan, dst_mp: Path | None, src_root: Path | None
    ) -> Callable[[], Destination]:
        if plan.dst_mount_id is None:
            root = Path(safe_abspath("dst_path", plan.dst_path))
@@ -263,7 +284,7 @@ class RuleRunner:
            kind = "cifs" if plan.dst_type is MountType.CIFS else "nfs"
            create = True

-        if local_root.resolve() == src_root.resolve():
+        if src_root is not None and local_root.resolve() == src_root.resolve():
            raise ValidationError("dst_path", "Quelle und Ziel sind derselbe Ordner")

        def make_local() -> Destination:
@@ -34,7 +34,7 @@ class FileState:
@dataclass(frozen=True)
 class Candidate:
    relpath: str  # POSIX, relativ zum Quellordner
-    path: Path
+    path: Path | None  # nur bei lokalen Quellen
    state: FileState


@@ -0,0 +1,311 @@
+"""Quellen: eingehängtes Verzeichnis (SMB/NFS) oder SFTP.
+
+Die Engine greift nur über diese Schnittstelle auf die Quelle zu – scannen,
+frischen Zustand messen, lesen, löschen, in die Quarantäne verschieben. So gelten
+für SFTP-Quellen dieselben Garantien wie für eingehängte Freigaben.
+
+Adressierung: ``root`` ist die Wurzel der Anbindung, ``subdir`` der Quell-
+Unterordner der Regel. Kandidaten tragen Pfade relativ zu ``subdir`` (so landen sie
+im Ziel und in der Buchführung); die Quarantäne liegt relativ zu ``root``.
+
+SFTP-Besonderheit: Änderungszeiten kommen nur sekundengenau. Die Stabilitäts-
+prüfung stützt sich dort vor allem auf die Größe – mit einem Fenster ab 2 s
+unproblematisch.
+"""
+
+from __future__ import annotations
+
+import contextlib
+import errno
+import os
+import posixpath
+import socket
+import stat
+from abc import ABC, abstractmethod
+from collections.abc import Iterator
+from dataclasses import dataclass
+from pathlib import Path
+from typing import IO, Any
+
+import paramiko
+
+from netmover.mounting.sftp import SftpSession
+from netmover.mover.fsops import rename_noreplace
+from netmover.mover.scan import (
+    IGNORED_DIRS,
+    Candidate,
+    FileState,
+    Matcher,
+    current_state,
+    is_ignored_name,
+    scan,
+)
+
+_NET_ERRORS = (paramiko.SSHException, EOFError, socket.timeout, ConnectionError)
+
+
+class SourceMissing(Exception):
+    """Quellordner existiert nicht (Lauf abbrechen, nichts anfassen)."""
+
+
+@dataclass(frozen=True)
+class SourceStat:
+    size: int
+    mtime_ns: int
+    atime_ns: int
+    mode: int  # Rechte-Bits
+
+
+def _join(*parts: str) -> str:
+    return posixpath.join(*[p for p in parts if p]) if any(parts) else ""
+
+
+@contextlib.contextmanager
+def network_errors() -> Iterator[None]:
+    """Verbindungsabbrüche bei SFTP als ``ENOTCONN`` melden – die Engine bricht den
+    Lauf dann ab (Anbindung gestört), statt jede Datei einzeln scheitern zu lassen."""
+    try:
+        yield
+    except _NET_ERRORS as exc:
+        raise OSError(errno.ENOTCONN, f"SFTP-Verbindung unterbrochen: {exc}") from exc
+
+
+class Source(ABC):
+    kind: str
+
+    def __init__(self, subdir: str = "") -> None:
+        self.subdir = subdir.strip("/")
+
+    def full(self, rel: str) -> str:
+        """Pfad relativ zur Wurzel der Anbindung."""
+        return _join(self.subdir, rel)
+
+    @abstractmethod
+    def describe(self, rel: str = "") -> str: ...
+
+    @abstractmethod
+    def check_root(self) -> None:
+        """Wirft :class:`SourceMissing`, wenn der Quellordner fehlt."""
+
+    @abstractmethod
+    def scan(
+        self, matcher: Matcher, recursive: bool, exclude: frozenset[str] = frozenset()
+    ) -> list[Candidate]:
+        """``exclude``: Verzeichnisse relativ zur Wurzel der Anbindung."""
+
+    @abstractmethod
+    def state(self, rel: str) -> FileState | None:
+        """Frischer Zustand; ``None`` = verschwunden, ``BUSY`` = gerade gesperrt."""
+
+    @abstractmethod
+    def stat(self, rel: str) -> SourceStat: ...
+
+    @abstractmethod
+    def open(self, rel: str) -> contextlib.AbstractContextManager[IO[bytes]]: ...
+
+    @abstractmethod
+    def delete(self, rel: str) -> None: ...
+
+    @abstractmethod
+    def exists_root_rel(self, root_rel: str) -> bool: ...
+
+    @abstractmethod
+    def describe_root_rel(self, root_rel: str) -> str: ...
+
+    @abstractmethod
+    def move_to(self, rel: str, root_rel_target: str) -> None:
+        """Innerhalb der Anbindung verschieben, ohne zu überschreiben
+        (``FileExistsError``, wenn das Ziel existiert)."""
+
+    def local_path(self, rel: str) -> Path | None:
+        """Lokaler Pfad, falls vorhanden (für den Schnellweg ohne Kopie)."""
+        return None
+
+    def close(self) -> None:  # noqa: B027 – absichtlich optional
+        pass
+
+
+class LocalSource(Source):
+    """Eingehängte Freigabe (oder lokales Verzeichnis)."""
+
+    kind = "local"
+
+    def __init__(self, root: Path, subdir: str = "") -> None:
+        super().__init__(subdir)
+        self.root = root
+        self.base = root / self.subdir if self.subdir else root
+        # zusätzlich auszulassende Verzeichnisse, z. B. ein Ziel innerhalb der Quelle
+        self.exclude_paths: set[Path] = set()
+
+    def _p(self, rel: str) -> Path:
+        return self.base / rel
+
+    def describe(self, rel: str = "") -> str:
+        return str(self._p(rel)) if rel else str(self.base)
+
+    def check_root(self) -> None:
+        if not self.base.is_dir():
+            raise SourceMissing(f"Quellordner {self.base} existiert nicht")
+
+    def scan(
+        self, matcher: Matcher, recursive: bool, exclude: frozenset[str] = frozenset()
+    ) -> list[Candidate]:
+        paths = {(self.root / e).resolve() for e in exclude} | set(self.exclude_paths)
+        return list(scan(self.base, matcher, recursive, frozenset(paths)))
+
+    def state(self, rel: str) -> FileState | None:
+        return current_state(self._p(rel))
+
+    def stat(self, rel: str) -> SourceStat:
+        st = os.stat(self._p(rel))
+        return SourceStat(st.st_size, st.st_mtime_ns, st.st_atime_ns, stat.S_IMODE(st.st_mode))
+
+    def open(self, rel: str) -> contextlib.AbstractContextManager[IO[bytes]]:
+        return self._p(rel).open("rb")
+
+    def delete(self, rel: str) -> None:
+        os.unlink(self._p(rel))
+
+    def exists_root_rel(self, root_rel: str) -> bool:
+        return os.path.lexists(self.root / root_rel)
+
+    def describe_root_rel(self, root_rel: str) -> str:
+        return str(self.root / root_rel)
+
+    def move_to(self, rel: str, root_rel_target: str) -> None:
+        target = self.root / root_rel_target
+        target.parent.mkdir(mode=0o755, parents=True, exist_ok=True)
+        if rename_noreplace(self._p(rel), target):
+            return
+        if os.path.lexists(target):
+            raise FileExistsError(errno.EEXIST, "exists", str(target))
+        os.rename(self._p(rel), target)
+
+    def local_path(self, rel: str) -> Path:
+        return self._p(rel)
+
+
+class SftpSource(Source):
+    """Quelle auf einem SFTP-Server (z. B. ein Server, der nur per SSH erreichbar ist)."""
+
+    kind = "sftp"
+
+    def __init__(self, session: SftpSession, base: str, subdir: str = "", label: str = ""):
+        super().__init__(subdir)
+        self.session = session
+        self.client = session.client
+        self.root = base.rstrip("/") or "/"
+        self.label = label or "sftp"
+
+    def _p(self, rel: str) -> str:
+        return posixpath.join(self.root, self.full(rel)) if self.full(rel) else self.root
+
+    def describe(self, rel: str = "") -> str:
+        return f"{self.label}:{self._p(rel)}"
+
+    def close(self) -> None:
+        self.session.close()
+
+    def check_root(self) -> None:
+        with network_errors():
+            try:
+                attrs = self.client.stat(self._p(""))
+            except FileNotFoundError:
+                raise SourceMissing(f"Quellordner {self.describe()} existiert nicht") from None
+        if attrs.st_mode is None or not stat.S_ISDIR(attrs.st_mode):
+            raise SourceMissing(f"{self.describe()} ist kein Verzeichnis")
+
+    def scan(
+        self, matcher: Matcher, recursive: bool, exclude: frozenset[str] = frozenset()
+    ) -> list[Candidate]:
+        excluded = {e.strip("/") for e in exclude}
+        found: list[Candidate] = []
+
+        def walk(rel_dir: str) -> None:
+            with network_errors():
+                entries = sorted(
+                    self.client.listdir_attr(self._p(rel_dir)), key=lambda a: a.filename
+                )
+            for a in entries:
+                name = a.filename
+                if name.startswith(".") or a.st_mode is None:
+                    continue
+                rel = _join(rel_dir, name)
+                if stat.S_ISDIR(a.st_mode):
+                    if recursive and name not in IGNORED_DIRS and self.full(rel) not in excluded:
+                        walk(rel)
+                    continue
+                if not stat.S_ISREG(a.st_mode) or is_ignored_name(name):
+                    continue  # Symlinks, Geräte … und unfertige Übertragungen
+                if matcher(rel):
+                    found.append(Candidate(rel, None, _state(a)))
+
+        walk("")
+        return found
+
+    def state(self, rel: str) -> FileState | None:
+        with network_errors():
+            try:
+                a = self.client.lstat(self._p(rel))
+            except FileNotFoundError:
+                return None
+        if a.st_mode is None or not stat.S_ISREG(a.st_mode):
+            return None
+        return _state(a)
+
+    def stat(self, rel: str) -> SourceStat:
+        with network_errors():
+            a = self.client.lstat(self._p(rel))
+        mtime = int(a.st_mtime or 0) * 1_000_000_000
+        atime = int(a.st_atime or a.st_mtime or 0) * 1_000_000_000
+        return SourceStat(int(a.st_size or 0), mtime, atime, stat.S_IMODE(a.st_mode or 0o644))
+
+    @contextlib.contextmanager
+    def _reader(self, rel: str) -> Iterator[IO[bytes]]:
+        with network_errors():
+            size = self.client.lstat(self._p(rel)).st_size
+            fh: Any = self.client.open(self._p(rel), "rb")
+            try:
+                fh.prefetch(size)  # Größe mitgeben: spart ein fstat und liest vorab
+                yield fh
+            finally:
+                with contextlib.suppress(*_NET_ERRORS, OSError):
+                    fh.close()
+
+    def open(self, rel: str) -> contextlib.AbstractContextManager[IO[bytes]]:
+        return self._reader(rel)
+
+    def delete(self, rel: str) -> None:
+        with network_errors():
+            self.client.remove(self._p(rel))
+
+    def exists_root_rel(self, root_rel: str) -> bool:
+        with network_errors():
+            try:
+                self.client.lstat(posixpath.join(self.root, root_rel))
+            except FileNotFoundError:
+                return False
+        return True
+
+    def describe_root_rel(self, root_rel: str) -> str:
+        return f"{self.label}:{posixpath.join(self.root, root_rel)}"
+
+    def move_to(self, rel: str, root_rel_target: str) -> None:
+        target = posixpath.join(self.root, root_rel_target)
+        with network_errors():
+            current = self.root
+            for part in posixpath.dirname(root_rel_target).split("/"):
+                if not part:
+                    continue
+                current = posixpath.join(current, part)
+                try:
+                    self.client.stat(current)
+                except FileNotFoundError:
+                    self.client.mkdir(current, mode=0o755)
+            if self.exists_root_rel(root_rel_target):
+                raise FileExistsError(errno.EEXIST, "exists", target)
+            self.client.rename(self._p(rel), target)  # SFTPv3-rename überschreibt nicht
+
+
+def _state(a: paramiko.SFTPAttributes) -> FileState:
+    return FileState(int(a.st_size or 0), int(a.st_mtime or 0) * 1_000_000_000)
@@ -46,6 +46,9 @@ class AppSettings:
    smtp_password_enc: str = ""
    smtp_from: str = ""
    smtp_to: str = ""
+    # Anmeldung: "always" | "trusted" (ohne Anmeldung aus trusted_networks) | "off"
+    auth_mode: str = "always"
+    trusted_networks: str = ""


 _PREFIX = "app."
@@ -49,8 +49,17 @@ from netmover.admin import (
 )
 from netmover.app_state import NetMover
 from netmover.config import Settings, get_settings
+from netmover.credentials import (
+    create_password,
+    delete_credential,
+    generate_key,
+    import_key,
+    list_keys,
+    list_passwords,
+    rename_credential,
+    update_password,
+)
 from netmover.db import session_scope
-from netmover.keys import delete_key, generate_key, import_key, list_keys, rename_key
 from netmover.kuma import push as kuma_push
 from netmover.logging_setup import event, get_logger
 from netmover.mail import SECURITY_MODES, MailConfigError, parse_recipients, valid_address
@@ -171,7 +180,7 @@ def flash(request: Request, message: str, kind: str = "ok") -> None:
 HELP_ANCHORS = {
    "status": "status",
    "mounts": "anbindungen",
-    "keys": "schluessel",
+    "keys": "zugangsdaten",
    "rules": "regeln",
    "log": "protokoll",
    "settings": "einstellungen",
@@ -191,7 +200,7 @@ class NavState:


 def _nav_state(request: Request) -> NavState | None:
-    if not auth.is_logged_in(request):
+    if not _authorized(request):
        return None
    state = nm(request)
    health = state.health()
@@ -228,7 +237,9 @@ def render(request: Request, name: str, status_code: int = 200, **ctx: Any) -> H
        {
            "csrf": auth.csrf_token(request),
            "flashes": flashes,
-            "logged_in": auth.is_logged_in(request),
+            "logged_in": _authorized(request),
+            "session_login": auth.is_logged_in(request),
+            "auth_bypass": getattr(request.state, "auth_bypass", None),
            "active": ctx.pop("active", ""),
            **ctx,
        },
@@ -254,13 +265,22 @@ def _form_dict(form: Any) -> dict[str, str]:

 async def guard(request: Request) -> None:
    state = nm(request)
+    # Ein Passwort ist immer Pflicht – sonst ließe sich die Anmeldung nie wieder einschalten.
    if not auth.password_is_set(state.session_factory):
        raise auth.LoginRequired(setup=True)
-    if not auth.is_logged_in(request):
+    cfg = load_settings(state.session_factory)
+    bypass = auth.login_bypass(request, cfg.auth_mode, cfg.trusted_networks)
+    if bypass:
+        request.state.auth_bypass = bypass
+    elif not auth.is_logged_in(request):
        raise auth.LoginRequired()
    await auth.check_csrf(request)


+def _authorized(request: Request) -> bool:
+    return auth.is_logged_in(request) or bool(getattr(request.state, "auth_bypass", None))
+
+
 protected = APIRouter(dependencies=[Depends(guard)])
 public = APIRouter()

@@ -340,8 +360,12 @@ async def setup_submit(request: Request) -> Response:

@public.get("/login", response_class=HTMLResponse)
 async def login_form(request: Request) -> Response:
-    if not auth.password_is_set(nm(request).session_factory):
+    state = nm(request)
+    if not auth.password_is_set(state.session_factory):
        return redirect("/setup")
+    cfg = load_settings(state.session_factory)
+    if auth.login_bypass(request, cfg.auth_mode, cfg.trusted_networks):
+        return redirect("/")
    return render(request, "login.html")


@@ -511,7 +535,7 @@ async def mounts_page(request: Request) -> Response:
            m.id: m.credential.name
            for m in mounts
            if m.credential is not None and m.credential.managed
-        }
+        }  # Name gespeicherter Zugangsdaten (Passwort oder Schlüssel)
        return render(
            request,
            "mounts.html",
@@ -533,6 +557,15 @@ def _report_data(raw: str | None) -> dict[str, Any]:
    return data


+def _password_options(request: Request) -> list[tuple[str, str]]:
+    with nm(request).session_factory() as s:
+        stored = list_passwords(s)
+    return [("own", "eigene Angaben für diese Anbindung")] + [
+        (str(p.id), f"{p.name} ({p.domain + chr(92) if p.domain else ''}{p.username})")
+        for p in stored
+    ]
+
+
 def _key_options(request: Request) -> list[tuple[str, str]]:
    with nm(request).session_factory() as s:
        keys = list_keys(s)
@@ -566,6 +599,7 @@ def _mount_form(
            mount is not None and mount.credential is not None and not mount.credential.managed
        ),
        key_options=_key_options(request),
+        password_options=_password_options(request),
    )


@@ -702,26 +736,75 @@ async def mount_reset_hostkey(request: Request, mount_id: int) -> Response:
    return redirect(f"/mounts#mount-{mount_id}")


-# -- Schlüsselverwaltung ----------------------------------------------------------------------
+# -- Zugangsdaten ------------------------------------------------------------------------------


-@protected.get("/keys", response_class=HTMLResponse)
-async def keys_page(request: Request) -> Response:
+@protected.get("/credentials", response_class=HTMLResponse)
+async def credentials_page(request: Request) -> Response:
    with nm(request).session_factory() as s:
-        return render(request, "keys.html", active="keys", keys=list_keys(s))
+        return render(
+            request,
+            "credentials.html",
+            active="keys",
+            keys=list_keys(s),
+            passwords=list_passwords(s),
+        )


-async def _key_action(request: Request, work: Callable[[Any], str]) -> Response:
+@protected.get("/keys")
+async def keys_redirect(request: Request) -> Response:
+    return RedirectResponse("/credentials", status_code=301)
+
+
+async def _cred_action(request: Request, work: Callable[[Any], str]) -> Response:
    state = nm(request)
    try:
        with session_scope(state.session_factory) as s:
            flash(request, work(s))
    except AdminError as exc:
        flash(request, str(exc), "error")
-    return redirect("/keys")
+    return redirect("/credentials")


-@protected.post("/keys/generate")
+@protected.post("/credentials/passwords")
+async def password_create(request: Request) -> Response:
+    form = _form_dict(await request.form())
+    secrets = nm(request).secrets
+
+    def work(s: Any) -> str:
+        cred = create_password(
+            s,
+            secrets,
+            form.get("name", ""),
+            form.get("username", ""),
+            form.get("password", ""),
+            form.get("domain", ""),
+        )
+        return f"Zugangsdaten „{cred.name}“ gespeichert."
+
+    return await _cred_action(request, work)
+
+
+@protected.post("/credentials/{cred_id}/password")
+async def password_update(request: Request, cred_id: int) -> Response:
+    form = _form_dict(await request.form())
+    secrets = nm(request).secrets
+
+    def work(s: Any) -> str:
+        cred = update_password(
+            s,
+            secrets,
+            cred_id,
+            form.get("username", ""),
+            form.get("password", ""),
+            form.get("domain", ""),
+        )
+        return f"Zugangsdaten „{cred.name}“ geändert – gilt ab sofort für alle Anbindungen."
+
+    return await _cred_action(request, work)
+
+
+@protected.post("/credentials/keys/generate")
 async def key_generate(request: Request) -> Response:
    form = _form_dict(await request.form())
    secrets = nm(request).secrets
@@ -730,10 +813,10 @@ async def key_generate(request: Request) -> Response:
        key = generate_key(s, secrets, form.get("name", ""))
        return f"Schlüssel „{key.name}“ erzeugt."

-    return await _key_action(request, work)
+    return await _cred_action(request, work)


-@protected.post("/keys/import")
+@protected.post("/credentials/keys/import")
 async def key_import(request: Request) -> Response:
    form = _form_dict(await request.form())
    secrets = nm(request).secrets
@@ -748,26 +831,26 @@ async def key_import(request: Request) -> Response:
        )
        return f"Schlüssel „{key.name}“ importiert."

-    return await _key_action(request, work)
+    return await _cred_action(request, work)


-@protected.post("/keys/{key_id}/rename")
-async def key_rename(request: Request, key_id: int) -> Response:
+@protected.post("/credentials/{cred_id}/rename")
+async def credential_rename(request: Request, cred_id: int) -> Response:
    form = _form_dict(await request.form())

    def work(s: Any) -> str:
-        rename_key(s, key_id, form.get("name", ""))
-        return "Schlüssel umbenannt."
+        rename_credential(s, cred_id, form.get("name", ""))
+        return "Umbenannt."

-    return await _key_action(request, work)
+    return await _cred_action(request, work)


-@protected.post("/keys/{key_id}/delete")
-async def key_delete(request: Request, key_id: int) -> Response:
+@protected.post("/credentials/{cred_id}/delete")
+async def credential_delete(request: Request, cred_id: int) -> Response:
    def work(s: Any) -> str:
-        return f"Schlüssel „{delete_key(s, key_id)}“ gelöscht."
+        return f"„{delete_credential(s, cred_id)}“ gelöscht."

-    return await _key_action(request, work)
+    return await _cred_action(request, work)


 # -- Regeln -----------------------------------------------------------------------------------
@@ -792,9 +875,8 @@ def _rule_form(
        d=d,
        errors=errors,
        rule_id=rule_id,
-        src_mounts=[m for m in mounts if m.type.is_kernel_mount],
-        src_options=[("", "– wählen –")]
-        + [(m.id, m.name) for m in mounts if m.type.is_kernel_mount],
+        src_mounts=mounts,
+        src_options=[("", "– wählen –")] + [(m.id, f"{m.name} ({m.type.value})") for m in mounts],
        dst_options=[("", "– wählen –")] + [(m.id, f"{m.name} ({m.type.value})") for m in mounts],
    )

@@ -994,6 +1076,8 @@ async def settings_page(request: Request) -> Response:
        cfg=cfg,
        errors={},
        monitor_urls=_monitor_urls(request),
+        client_ip=request.client.host if request.client else "",
+        loopback_trusted=auth.includes_loopback(cfg.trusted_networks),
    )


@@ -1149,6 +1233,43 @@ async def settings_password(request: Request) -> Response:
    return redirect("/settings")


+@protected.post("/settings/auth", response_class=HTMLResponse)
+async def settings_auth(request: Request) -> Response:
+    """Anmeldung abschalten bzw. auf vertrauenswürdige Netze beschränken.
+    Nur mit dem aktuellen Passwort."""
+    state = nm(request)
+    form = _form_dict(await request.form())
+    if not await run_in_threadpool(
+        auth.verify_password, state.session_factory, form.get("current", "")
+    ):
+        flash(request, "Aktuelles Passwort falsch – Anmeldung unverändert.", "error")
+        return redirect("/settings#anmeldung")
+    mode = form.get("auth_mode", auth.AUTH_ALWAYS)
+    networks = (form.get("trusted_networks") or "").strip()
+    if mode not in auth.AUTH_MODES:
+        flash(request, "Ungültige Auswahl.", "error")
+        return redirect("/settings#anmeldung")
+    try:
+        nets = auth.parse_networks(networks)
+    except ValueError as exc:
+        flash(request, str(exc), "error")
+        return redirect("/settings#anmeldung")
+    if mode == auth.AUTH_TRUSTED and not nets:
+        flash(request, "Bitte mindestens ein vertrauenswürdiges Netz angeben.", "error")
+        return redirect("/settings#anmeldung")
+    cfg = load_settings(state.session_factory)
+    cfg.auth_mode, cfg.trusted_networks = mode, networks
+    save_settings(state.session_factory, cfg)
+    event(log, "auth.mode_changed", level=30, mode=mode, networks=networks)
+    msgs = {
+        auth.AUTH_ALWAYS: "Anmeldung ist wieder immer erforderlich.",
+        auth.AUTH_TRUSTED: f"Ohne Anmeldung aus: {networks}. Von überall sonst mit Passwort.",
+        auth.AUTH_OFF: "Anmeldung abgeschaltet – NetMover ist ohne Passwort bedienbar.",
+    }
+    flash(request, msgs[mode], "info" if mode == auth.AUTH_ALWAYS else "warn")
+    return redirect("/settings#anmeldung")
+
+
@protected.get("/settings/export")
 async def settings_export(request: Request) -> Response:
    state = nm(request)
@@ -12,6 +12,8 @@
 from __future__ import annotations

 import hmac
+import ipaddress
+import re
 import secrets
 import threading
 import time
@@ -32,6 +34,54 @@ SAFE_METHODS = frozenset({"GET", "HEAD", "OPTIONS"})
 _hasher = PasswordHasher()


+AUTH_ALWAYS, AUTH_TRUSTED, AUTH_OFF = "always", "trusted", "off"
+AUTH_MODES = (AUTH_ALWAYS, AUTH_TRUSTED, AUTH_OFF)
+
+
+def parse_networks(raw: str) -> list[ipaddress.IPv4Network | ipaddress.IPv6Network]:
+    """„10.0.10.0/24, 192.168.1.5“ → Netze; ``ValueError`` mit verständlicher Meldung."""
+    nets = []
+    for part in re.split(r"[,;\s]+", raw.strip()):
+        if not part:
+            continue
+        try:
+            nets.append(ipaddress.ip_network(part, strict=False))
+        except ValueError:
+            raise ValueError(f"„{part}“ ist kein gültiges Netz (z. B. 10.0.10.0/24)") from None
+    return nets
+
+
+def login_bypass(request: Request, mode: str, trusted: str) -> str | None:
+    """Grund, warum keine Anmeldung nötig ist (``"off"``/``"trusted"``), sonst ``None``.
+
+    Hinter einem Reverse Proxy ist die Client-Adresse nur dann echt, wenn
+    ``NETMOVER_FORWARDED_ALLOW_IPS`` den Proxy kennt – sonst sähe jede Anfrage aus,
+    als käme sie vom Proxy selbst."""
+    if mode == AUTH_OFF:
+        return AUTH_OFF
+    if mode != AUTH_TRUSTED or request.client is None:
+        return None
+    try:
+        addr = ipaddress.ip_address(request.client.host)
+    except ValueError:
+        return None
+    try:
+        nets = parse_networks(trusted)
+    except ValueError:
+        return None
+    return AUTH_TRUSTED if any(addr in n for n in nets) else None
+
+
+def includes_loopback(raw: str) -> bool:
+    try:
+        nets = parse_networks(raw)
+    except ValueError:
+        return False
+    return any(
+        ipaddress.ip_address("127.0.0.1") in n or ipaddress.ip_address("::1") in n for n in nets
+    )
+
+
 class LoginRequired(Exception):
    def __init__(self, setup: bool = False) -> None:
        self.setup = setup
@@ -23,6 +23,7 @@
  {% if nav %}
  <a class="health {{ 'ok' if nav.healthy else 'bad' }}" href="/" title="Gesundheitszustand">
    <span class="dot"></span>{{ 'Alles in Ordnung' if nav.healthy else (nav.problems ~ (' Problem' if nav.problems == 1 else ' Probleme')) }}</a>
+  {% if auth_bypass %}<a class="health warn" href="/settings#anmeldung" title="Ohne Passwort bedienbar">{{ 'ohne Anmeldung' if auth_bypass == 'off' else 'ohne Anmeldung (vertrautes Netz)' }}</a>{% endif %}
  {% if nav.maintenance %}<a class="health warn" href="/" title="Geplante Läufe sind pausiert"><span class="dot"></span>Wartungsmodus</a>{% endif %}
  {% endif %}
  <span class="spacer"></span>
@@ -36,7 +37,7 @@
  <div class="nav-group">
    <div class="nav-title">Verbindungen</div>
    {{ item('mounts', '/mounts', 'Anbindungen', (nav.mounts_bad ~ ' gestört') if nav and nav.mounts_bad else (nav.mounts if nav and nav.mounts else ''), 'bad' if nav and nav.mounts_bad else '') }}
-    {{ item('keys', '/keys', 'Schlüssel', nav.keys if nav and nav.keys else '') }}
+    {{ item('keys', '/credentials', 'Zugangsdaten', nav.keys if nav and nav.keys else '') }}
  </div>
  <div class="nav-group">
    <div class="nav-title">Automatisierung</div>
@@ -49,10 +50,12 @@
    {{ item('help', '/help', 'Hilfe') }}
  </div>
  <div class="nav-foot">
+    {% if session_login %}
    <form method="post" action="/logout">
      <input type="hidden" name="csrf_token" value="{{ csrf }}">
      <button class="nav-button">{{ icon('logout') }}<span>Abmelden</span></button>
    </form>
+    {% endif %}
    <div class="version">NetMover {{ version }}</div>
  </div>
 </nav>
@@ -0,0 +1,97 @@
+{% extends "base.html" %}
+{% from "macros.html" import csrf_field %}
+{% block title %}Zugangsdaten{% endblock %}
+{% block content %}
+<h1>Zugangsdaten</h1>
+<p class="muted small">Einmal hinterlegen, in beliebig vielen Anbindungen auswählen. Passwörter und private Schlüssel
+  liegen verschlüsselt in der Datenbank und werden nie angezeigt.</p>
+
+<h2>Passwörter <a class="legend-help" href="/help#zugangsdaten" title="Hilfe">?</a></h2>
+<p class="muted small">Für SMB-Freigaben und SFTP-Server mit Passwort. Ein neues Passwort gilt sofort für alle Anbindungen, die es nutzen.</p>
+{% if not passwords %}<p class="muted">Noch keine Passwörter gespeichert.</p>{% endif %}
+<div class="stack">
+{% for p in passwords %}
+<div class="card" id="cred-{{ p.id }}">
+  <div class="row spread">
+    <div><strong>{{ p.name }}</strong> <span class="badge">{% if p.domain %}{{ p.domain }}\{% endif %}{{ p.username }}</span>
+      {% if p.used_by %}<span class="badge ok">benutzt von {{ p.used_by|join(', ') }}</span>{% else %}<span class="badge">unbenutzt</span>{% endif %}</div>
+    <div class="row">
+      <details class="rename"><summary class="btn small">Ändern</summary>
+        <form class="stack" method="post" action="/credentials/{{ p.id }}/password">{{ csrf_field(csrf) }}
+          <input type="text" name="username" value="{{ p.username }}" aria-label="Benutzer" required>
+          <input type="text" name="domain" value="{{ p.domain or '' }}" placeholder="Domäne (optional)" aria-label="Domäne">
+          <input type="password" name="password" placeholder="neues Passwort (leer = unverändert)" aria-label="Passwort" autocomplete="new-password">
+          <button class="small primary">Speichern</button></form></details>
+      <details class="rename"><summary class="btn small">Umbenennen</summary>
+        <form class="row" method="post" action="/credentials/{{ p.id }}/rename">{{ csrf_field(csrf) }}
+          <input type="text" name="name" value="{{ p.name }}" aria-label="Neuer Name" required maxlength="100">
+          <button class="small primary">Speichern</button></form></details>
+      <form class="inline" method="post" action="/credentials/{{ p.id }}/delete" data-confirm="„{{ p.name }}“ löschen?">{{ csrf_field(csrf) }}
+        <button class="small danger"{% if p.used_by %} disabled title="Wird noch benutzt"{% endif %}>Löschen</button></form>
+    </div>
+  </div>
+  <div class="small muted">angelegt {{ p.created_at|localdt }}</div>
+</div>
+{% endfor %}
+</div>
+<form class="form card stack" method="post" action="/credentials/passwords">{{ csrf_field(csrf) }}
+  <div class="fields">
+    <div class="field"><label for="p-name">Name</label><input id="p-name" type="text" name="name" placeholder="z. B. NAS-Scanner" required maxlength="100"></div>
+    <div class="field"><label for="p-user">Benutzer</label><input id="p-user" type="text" name="username" required autocomplete="off"></div>
+    <div class="field"><label for="p-pass">Passwort</label><input id="p-pass" type="password" name="password" required autocomplete="new-password"></div>
+    <div class="field"><label for="p-dom">Domäne (optional)</label><input id="p-dom" type="text" name="domain"></div>
+  </div>
+  <button class="primary">Passwort speichern</button>
+</form>
+
+<h2>SSH-Schlüssel</h2>
+<p class="muted small">Für SFTP-Anbindungen. Ein Schlüssel kann von mehreren Anbindungen benutzt werden; den Benutzernamen legt jede Anbindung selbst fest.</p>
+
+{% if not keys %}<p class="muted">Noch keine Schlüssel gespeichert.</p>{% endif %}
+<div class="stack">
+{% for k in keys %}
+<div class="card" id="key-{{ k.id }}">
+  <div class="row spread">
+    <div><strong>{{ k.name }}</strong> <span class="badge">{{ k.key_type.removeprefix('ssh-') }}</span>
+      {% if k.used_by %}<span class="badge ok">benutzt von {{ k.used_by|join(', ') }}</span>{% else %}<span class="badge">unbenutzt</span>{% endif %}</div>
+    <div class="row">
+      <details class="rename"><summary class="btn small">Umbenennen</summary>
+        <form class="row" method="post" action="/credentials/{{ k.id }}/rename">{{ csrf_field(csrf) }}
+          <input type="text" name="name" value="{{ k.name }}" aria-label="Neuer Name" required maxlength="100">
+          <button class="small primary">Speichern</button></form></details>
+      <form class="inline" method="post" action="/credentials/{{ k.id }}/delete" data-confirm="Schlüssel „{{ k.name }}“ löschen?">{{ csrf_field(csrf) }}
+        <button class="small danger"{% if k.used_by %} disabled title="Wird noch benutzt"{% endif %}>Löschen</button></form>
+    </div>
+  </div>
+  <div class="small muted">Fingerabdruck <span class="mono">{{ k.fingerprint }}</span> · angelegt {{ k.created_at|localdt }}</div>
+  <details class="small"><summary>Öffentlicher Schlüssel / authorized_keys</summary>
+    <div class="field"><label>Zeile für <code>~/.ssh/authorized_keys</code> (nur SFTP erlaubt, empfohlen)</label>
+      <pre class="mono card" id="auth-{{ k.id }}">{{ k.authorized_keys_line }}</pre>
+      <button class="small" type="button" data-copy="auth-{{ k.id }}">kopieren</button></div>
+    <div class="field"><label>Öffentlicher Schlüssel</label>
+      <pre class="mono card" id="pub-{{ k.id }}">{{ k.public_key }}</pre>
+      <button class="small" type="button" data-copy="pub-{{ k.id }}">kopieren</button></div>
+  </details>
+</div>
+{% endfor %}
+</div>
+
+<h3>Neuen Schlüssel erzeugen</h3>
+<form class="form card row" method="post" action="/credentials/keys/generate">{{ csrf_field(csrf) }}
+  <input class="w-auto" type="text" name="name" placeholder="Name, z. B. paperless-lxc" aria-label="Name" required maxlength="100">
+  <button class="primary">Ed25519-Schlüssel erzeugen</button>
+</form>
+
+<h3>Vorhandenen Schlüssel importieren</h3>
+<form class="form card stack" method="post" action="/credentials/keys/import">{{ csrf_field(csrf) }}
+  <div class="fields">
+    <div class="field"><label for="k-name">Name</label><input id="k-name" type="text" name="name" required maxlength="100"></div>
+    <div class="field"><label for="k-pass">Passphrase (falls vorhanden)</label><input id="k-pass" type="password" name="passphrase" autocomplete="off"></div>
+    <div class="field wide"><label for="k-priv">Privater Schlüssel</label>
+      <textarea id="k-priv" name="private_key" class="mono" rows="8" autocomplete="off" spellcheck="false"
+        placeholder="-----BEGIN OPENSSH PRIVATE KEY-----&#10;…&#10;-----END OPENSSH PRIVATE KEY-----" required></textarea>
+      <span class="hint">Ed25519, ECDSA oder RSA (mind. 2048 Bit), OpenSSH- oder PEM-Format. Eine Passphrase wird nur zum Entschlüsseln benutzt.</span></div>
+  </div>
+  <button>Importieren</button>
+</form>
+{% endblock %}
@@ -1,57 +0,0 @@
-{% extends "base.html" %}
-{% from "macros.html" import csrf_field %}
-{% block title %}Schlüssel{% endblock %}
-{% block content %}
-<h1>Schlüssel</h1>
-<p class="muted small">Gespeicherte Schlüssel für SFTP-Anbindungen. Ein Schlüssel kann von mehreren
-  Anbindungen benutzt werden; den Benutzernamen legt jede Anbindung selbst fest. Private Schlüssel
-  liegen verschlüsselt in der Datenbank und werden nie angezeigt.</p>
-
-{% if not keys %}<p class="muted">Noch keine Schlüssel gespeichert.</p>{% endif %}
-<div class="stack">
-{% for k in keys %}
-<div class="card" id="key-{{ k.id }}">
-  <div class="row spread">
-    <div><strong>{{ k.name }}</strong> <span class="badge">{{ k.key_type.removeprefix('ssh-') }}</span>
-      {% if k.used_by %}<span class="badge ok">benutzt von {{ k.used_by|join(', ') }}</span>{% else %}<span class="badge">unbenutzt</span>{% endif %}</div>
-    <div class="row">
-      <details class="rename"><summary class="btn small">Umbenennen</summary>
-        <form class="row" method="post" action="/keys/{{ k.id }}/rename">{{ csrf_field(csrf) }}
-          <input type="text" name="name" value="{{ k.name }}" aria-label="Neuer Name" required maxlength="100">
-          <button class="small primary">Speichern</button></form></details>
-      <form class="inline" method="post" action="/keys/{{ k.id }}/delete" data-confirm="Schlüssel „{{ k.name }}“ löschen?">{{ csrf_field(csrf) }}
-        <button class="small danger"{% if k.used_by %} disabled title="Wird noch benutzt"{% endif %}>Löschen</button></form>
-    </div>
-  </div>
-  <div class="small muted">Fingerabdruck <span class="mono">{{ k.fingerprint }}</span> · angelegt {{ k.created_at|localdt }}</div>
-  <details class="small"><summary>Öffentlicher Schlüssel / authorized_keys</summary>
-    <div class="field"><label>Zeile für <code>~/.ssh/authorized_keys</code> (nur SFTP erlaubt, empfohlen)</label>
-      <pre class="mono card" id="auth-{{ k.id }}">{{ k.authorized_keys_line }}</pre>
-      <button class="small" type="button" data-copy="auth-{{ k.id }}">kopieren</button></div>
-    <div class="field"><label>Öffentlicher Schlüssel</label>
-      <pre class="mono card" id="pub-{{ k.id }}">{{ k.public_key }}</pre>
-      <button class="small" type="button" data-copy="pub-{{ k.id }}">kopieren</button></div>
-  </details>
-</div>
-{% endfor %}
-</div>
-
-<h2>Neuen Schlüssel erzeugen</h2>
-<form class="form card row" method="post" action="/keys/generate">{{ csrf_field(csrf) }}
-  <input class="w-auto" type="text" name="name" placeholder="Name, z. B. paperless-lxc" aria-label="Name" required maxlength="100">
-  <button class="primary">Ed25519-Schlüssel erzeugen</button>
-</form>
-
-<h2>Vorhandenen Schlüssel importieren</h2>
-<form class="form card stack" method="post" action="/keys/import">{{ csrf_field(csrf) }}
-  <div class="fields">
-    <div class="field"><label for="k-name">Name</label><input id="k-name" type="text" name="name" required maxlength="100"></div>
-    <div class="field"><label for="k-pass">Passphrase (falls vorhanden)</label><input id="k-pass" type="password" name="passphrase" autocomplete="off"></div>
-    <div class="field wide"><label for="k-priv">Privater Schlüssel</label>
-      <textarea id="k-priv" name="private_key" class="mono" rows="8" autocomplete="off" spellcheck="false"
-        placeholder="-----BEGIN OPENSSH PRIVATE KEY-----&#10;…&#10;-----END OPENSSH PRIVATE KEY-----" required></textarea>
-      <span class="hint">Ed25519, ECDSA oder RSA (mind. 2048 Bit), OpenSSH- oder PEM-Format. Eine Passphrase wird nur zum Entschlüsseln benutzt.</span></div>
-  </div>
-  <button>Importieren</button>
-</form>
-{% endblock %}
@@ -9,7 +9,7 @@
  {% if mount %}<input type="hidden" name="mount_id" value="{{ mount.id }}">{% endif %}
  <fieldset><legend>Verbindung <a class="legend-help" href="/help#anbindung-felder" title="Hilfe">?</a></legend><div class="fields">
    {{ field('name', 'Name', d.name, errors=errors, attrs='required maxlength="100"', hint=('Verzeichnis: ' ~ mount.mountpoint if mount and mount.mountpoint else '')) }}
-    {{ select('type', 'Typ', [('cifs', 'SMB/CIFS'), ('nfs4', 'NFS 4'), ('nfs3', 'NFS 3'), ('sftp', 'SFTP (nur Ziel)')], d.type.value, errors=errors) }}
+    {{ select('type', 'Typ', [('cifs', 'SMB/CIFS'), ('nfs4', 'NFS 4'), ('nfs3', 'NFS 3'), ('sftp', 'SFTP')], d.type.value, errors=errors) }}
    {{ field('server', 'Server', d.server, errors=errors, placeholder='192.168.1.10 oder nas.lan', attrs='required') }}
    <div class="field{{ ' has-error' if errors.get('share') }}">
      <label for="f-share">Freigabe / Pfad</label>
@@ -25,16 +25,21 @@
  <fieldset data-show-for="type:cifs sftp"><legend>Anmeldung <a class="legend-help" href="/help#anbindung-felder" title="Hilfe">?</a></legend><div class="fields">
    <div data-show-for="type:sftp">{{ select('auth', 'Anmeldeart', [('ssh_key', 'SSH-Schlüssel'), ('password', 'Passwort')], d.auth.value, errors=errors) }}</div>
    <div data-show-for="type:sftp"><div data-show-for="auth:ssh_key">
-      {{ select('key_id', 'Schlüssel', key_options, d.key_id, errors=errors, hint='Gespeicherte Schlüssel verwalten unter „Schlüssel“.') }}
+      {{ select('key_id', 'Schlüssel', key_options, d.key_id, errors=errors, hint='Verwalten unter „Zugangsdaten“.') }}
    </div></div>
+    <div data-show-for="type:cifs|auth:password">
+      {{ select('cred_id', 'Zugangsdaten', password_options, d.cred_id, errors=errors, hint='Gespeicherte Passwörter verwalten unter „Zugangsdaten“.') }}
+    </div>
+    <div data-show-for="auth:ssh_key|cred_id:own">
      {{ field('username', 'Benutzername', d.username, errors=errors, attrs='autocomplete="off"') }}
-    <div data-show-for="type:cifs">{{ field('domain', 'Domäne (optional)', d.domain, errors=errors) }}</div>
-    <div class="field{{ ' has-error' if errors.get('password') }}" id="pw-field" data-show-for="type:cifs|auth:password">
+    </div>
+    <div data-show-for="cred_id:own"><div data-show-for="type:cifs">{{ field('domain', 'Domäne (optional)', d.domain, errors=errors) }}</div></div>
+    <div data-show-for="cred_id:own"><div class="field{{ ' has-error' if errors.get('password') }}" id="pw-field" data-show-for="type:cifs|auth:password">
      <label for="f-password">Passwort</label>
      <input id="f-password" type="password" name="password" autocomplete="new-password" placeholder="{{ '••••••' if has_secret }}">
      {% if errors.get('password') %}<span class="error">{{ errors['password'] }}</span>{% endif %}
      <span class="hint">{{ 'Leer lassen = unverändert. ' if has_secret }}<span data-show-for="type:cifs">Benutzer und Passwort leer = Gastzugang.</span></span>
-    </div>
+    </div></div>
  </div>
  {% if public_key %}<div class="field wide"><label>Öffentlicher Schlüssel dieser Anbindung (in ~/.ssh/authorized_keys des Zielbenutzers eintragen)</label>
    <pre class="mono card" id="pubkey">{{ public_key }}</pre><button class="small" type="button" data-copy="pubkey">kopieren</button></div>{% endif %}
@@ -28,13 +28,11 @@
  <div class="muted small mono">{{ m.type.value }} · {% if m.type == MountType.CIFS %}//{{ m.server }}/{{ m.share }}{% elif m.type == MountType.SFTP %}{{ m.server }}:{{ m.port or 22 }}{{ m.share }}{% else %}{{ m.server }}:{{ m.share }}{% endif %}
    {% if m.mountpoint %} → {{ m.mountpoint }}{% endif %}</div>
  {% if m.last_error %}<pre class="small text-err">{{ m.last_error }}</pre>{% endif %}
+  {% if key_names.get(m.id) %}<div class="small">Anmeldung mit gespeicherten Zugangsdaten <a href="/credentials">„{{ key_names[m.id] }}“</a></div>{% endif %}
  {% if m.type == MountType.SFTP %}
    <div class="small">Hostschlüssel: {% if m.host_key %}<span class="mono">{{ m.host_key.split()[0] }}</span> festgehalten
      <form class="inline" method="post" action="/mounts/{{ m.id }}/reset-hostkey" data-confirm="Hostschlüssel zurücksetzen? Nur tun, wenn der Server neu aufgesetzt wurde.">{{ csrf_field(csrf) }}<button class="small">zurücksetzen</button></form>
      {% else %}<span class="text-warn">noch nicht bestätigt – bitte „Testen“</span>{% endif %}</div>
-    {% if key_names.get(m.id) %}
-    <div class="small">Anmeldung mit gespeichertem Schlüssel <a href="/keys">„{{ key_names[m.id] }}“</a></div>
-    {% endif %}
    {% if pubkeys.get(m.id) %}
    <details class="small"><summary>Öffentlicher Schlüssel für authorized_keys</summary>
      <pre class="mono" id="pub-{{ m.id }}">{{ pubkeys[m.id] }}</pre>
@@ -4,7 +4,7 @@
 {% block content %}
 <h1>{{ 'Regel bearbeiten' if rule_id else 'Neue Regel' }}</h1>
 {% if errors.get('_') %}<p class="msg error">{{ errors['_'] }}</p>{% endif %}
-{% if not src_mounts %}<p class="msg info">Zuerst eine SMB- oder NFS-Anbindung als Quelle <a href="/mounts/new">anlegen</a>.</p>{% endif %}
+{% if not src_mounts %}<p class="msg info">Zuerst eine Anbindung als Quelle <a href="/mounts/new">anlegen</a>.</p>{% endif %}
 <form class="form" method="post" data-dynamic>
  <input type="hidden" name="csrf_token" value="{{ csrf }}">
  <fieldset><legend>Allgemein <a class="legend-help" href="/help#regeln" title="Hilfe">?</a></legend><div class="fields">
@@ -65,6 +65,24 @@
  <button>Passwort ändern</button>
 </form>

+<h2 id="anmeldung">Anmeldung <a class="legend-help" href="/help#anmeldung" title="Hilfe">?</a></h2>
+<form class="form card stack" method="post" action="/settings/auth" data-dynamic>
+  {{ csrf_field(csrf) }}
+  <div class="fields">
+    {{ select('auth_mode', 'Anmeldung erforderlich', [('always', 'immer (empfohlen)'), ('trusted', 'nicht aus vertrauten Netzen'), ('off', 'nie')], cfg.auth_mode) }}
+    <div class="field wide" data-show-for="auth_mode:trusted"><label for="f-nets">Vertrauenswürdige Netze</label>
+      <input id="f-nets" type="text" name="trusted_networks" value="{{ cfg.trusted_networks }}" placeholder="10.0.10.0/24, 192.168.1.20">
+      <span class="hint">Mit Komma getrennt. Deine Adresse gerade: <code>{{ client_ip }}</code>. Von allen anderen Adressen wird weiter nach dem Passwort gefragt.</span></div>
+    <div class="field"><label for="f-cur-auth">Aktuelles Passwort (zur Bestätigung)</label>
+      <input id="f-cur-auth" type="password" name="current" autocomplete="current-password" required></div>
+  </div>
+  <p class="msg warn small" data-show-for="auth_mode:off|auth_mode:trusted">NetMover läuft als root und kennt die Zugangsdaten deiner Freigaben.
+    Ohne Anmeldung kann jeder, der die Oberfläche erreicht, Anbindungen und Regeln ändern. „Nie“ nur hinter einem Reverse Proxy mit eigener Anmeldung (z. B. Authelia) verwenden.
+    Hinter einem Proxy außerdem <code>NETMOVER_FORWARDED_ALLOW_IPS</code> setzen, sonst sieht jede Anfrage aus, als käme sie vom Proxy.</p>
+  {% if loopback_trusted %}<p class="msg warn small">Die Liste enthält 127.0.0.1/::1. Läuft ein Reverse Proxy auf diesem Server, gilt dann <strong>jede</strong> Anfrage über den Proxy als vertrauenswürdig.</p>{% endif %}
+  <div class="row"><button class="primary">Anmeldung speichern</button></div>
+</form>
+
 <h2>Export / Import</h2>
 <div class="card stack">
  <p class="small muted">Export enthält Anbindungen, Regeln und Einstellungen – <strong>ohne</strong> Zugangsdaten und Webhook-URL.
@@ -48,7 +48,11 @@ class _Server(paramiko.ServerInterface):


 class _Handle(SFTPHandle):
-    pass
+    def stat(self) -> Any:  # fstat wie bei OpenSSH
+        try:
+            return SFTPAttributes.from_stat(os.fstat(self.readfile.fileno()))
+        except OSError as e:
+            return SFTPServer.convert_errno(e.errno)


 class _RootedSFTP(SFTPServerInterface):
@@ -0,0 +1,114 @@
+"""0.5.0: Anmeldung abschaltbar – immer / nicht aus vertrauenswürdigen Netzen / nie."""
+
+from __future__ import annotations
+
+from typing import Any
+
+import pytest
+from fastapi.testclient import TestClient
+
+from netmover.settings_store import load_settings
+from netmover.web.auth import includes_loopback, parse_networks
+from tests.webutil import PASSWORD, csrf_of, post
+
+
+def _set_mode(client: TestClient, mode: str, nets: str = "", current: str = PASSWORD) -> Any:
+    return post(
+        client, "/settings/auth", {"auth_mode": mode, "trusted_networks": nets, "current": current}
+    )
+
+
+def _other(client: TestClient, host: str) -> TestClient:
+    """Weiterer Client für dieselbe App (ohne eigenen Lebenszyklus) mit fester Adresse."""
+    return TestClient(client.app, follow_redirects=False, client=(host, 51000))
+
+
+@pytest.fixture
+def lan_client(client: TestClient) -> TestClient:
+    return _other(client, "10.0.10.50")
+
+
+def test_default_requires_login(client: TestClient, lan_client: TestClient) -> None:
+    assert lan_client.get("/").status_code == 303
+
+
+def test_mode_change_needs_current_password(client: TestClient, state: Any) -> None:
+    _set_mode(client, "off", current="falsch")
+    assert load_settings(state.session_factory).auth_mode == "always"
+    assert "Aktuelles Passwort falsch" in client.get("/settings").text
+
+
+def test_trusted_networks(
+    client: TestClient, lan_client: TestClient, settings: Any, state: Any
+) -> None:
+    _set_mode(client, "trusted", "10.0.10.0/24")
+    r = lan_client.get("/")
+    assert r.status_code == 200 and "ohne Anmeldung (vertrautes Netz)" in r.text
+    assert "Abmelden" not in r.text
+    assert lan_client.get("/login").headers["location"] == "/"
+    # CSRF gilt weiterhin
+    assert lan_client.post("/maintenance").status_code == 403
+    token = csrf_of(lan_client.get("/").text)
+    assert lan_client.post("/maintenance", data={"csrf_token": token}).status_code == 303
+    # andere Netze brauchen weiter das Passwort
+    assert _other(client, "192.168.5.5").get("/").status_code == 303
+
+
+def test_off_and_back_on(client: TestClient, lan_client: TestClient) -> None:
+    _set_mode(client, "off")
+    r = lan_client.get("/rules")
+    assert r.status_code == 200 and "ohne Anmeldung" in r.text
+    token = csrf_of(r.text)
+    lan_client.post(
+        "/settings/auth",
+        data={
+            "csrf_token": token,
+            "auth_mode": "always",
+            "trusted_networks": "",
+            "current": PASSWORD,
+        },
+    )
+    assert lan_client.get("/").status_code == 303
+
+
+def test_setup_still_required_without_password(raw_client: TestClient, state: Any) -> None:
+    from netmover.settings_store import save_settings
+
+    cfg = load_settings(state.session_factory)
+    cfg.auth_mode = "off"
+    save_settings(state.session_factory, cfg)
+    assert raw_client.get("/").headers["location"] == "/setup"
+
+
+def test_validation(client: TestClient, state: Any) -> None:
+    _set_mode(client, "trusted", "")
+    assert "mindestens ein" in client.get("/settings").text
+    _set_mode(client, "trusted", "10.0.10.0/24, quatsch")
+    assert "kein gültiges Netz" in client.get("/settings").text
+    assert load_settings(state.session_factory).auth_mode == "always"
+    _set_mode(client, "trusted", "127.0.0.1")
+    assert "jede</strong> Anfrage über den Proxy" in client.get("/settings").text
+
+
+def test_network_parsing() -> None:
+    nets = parse_networks("10.0.10.0/24; 192.168.1.20 fe80::/10")
+    assert [str(n) for n in nets] == ["10.0.10.0/24", "192.168.1.20/32", "fe80::/10"]
+    assert includes_loopback("10.0.0.0/8, 127.0.0.0/8") and not includes_loopback("10.0.0.0/8")
+    with pytest.raises(ValueError):
+        parse_networks("10.0.0.0/33")
+
+
+def test_reset_password_turns_login_back_on(
+    state: Any, settings: Any, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    from netmover import cli
+    from netmover.settings_store import save_settings
+    from netmover.web.auth import set_password
+
+    set_password(state.session_factory, PASSWORD)
+    cfg = load_settings(state.session_factory)
+    cfg.auth_mode = "off"
+    save_settings(state.session_factory, cfg)
+    monkeypatch.setattr(cli, "get_settings", lambda: settings)
+    assert cli.main(["reset-password"]) == 0
+    assert load_settings(state.session_factory).auth_mode == "always"
@@ -13,9 +13,16 @@ from sqlalchemy import select
 from sqlalchemy.orm import Session, sessionmaker

 from netmover.admin import AdminError, MountInput, delete_mount, save_mount
+from netmover.credentials import (
+    delete_key,
+    fingerprint,
+    generate_key,
+    import_key,
+    list_keys,
+    rename_key,
+)
 from netmover.crypto import SecretBox, generate_ssh_keypair
 from netmover.db import session_scope
-from netmover.keys import delete_key, fingerprint, generate_key, import_key, list_keys, rename_key
 from netmover.models import Credential, CredentialKind, Mount, MountType
 from netmover.mounting.manager import MountConfig, MountManager
 from tests.sftp_stub import StubSFTPServer
@@ -142,7 +149,7 @@ def test_import_errors(session: Session, secretbox: SecretBox) -> None:

 def test_names(session: Session, secretbox: SecretBox) -> None:
    generate_key(session, secretbox, "eins")
-    with pytest.raises(AdminError, match="existiert bereits"):
+    with pytest.raises(AdminError, match="existieren bereits"):
        generate_key(session, secretbox, "eins")
    with pytest.raises(AdminError, match="reserviert"):
        generate_key(session, secretbox, "mount:x")
@@ -275,3 +282,102 @@ def test_list_keys_ignores_per_mount_credentials(
    save_mount(session, secretbox, manager, _sftp_input())
    assert list_keys(session) == []
    assert session.scalars(select(Credential)).one().managed is False
+
+
+# -- Gespeicherte Passwörter (0.5.0) ----------------------------------------------------------
+
+
+def _cifs(**kw: object) -> MountInput:
+    data: dict[str, object] = {
+        "name": "nas",
+        "type": MountType.CIFS,
+        "server": "10.0.10.6",
+        "share": "scans",
+    }
+    data.update(kw)
+    return MountInput(**data)  # type: ignore[arg-type]
+
+
+def test_stored_password_shared_by_mounts(
+    session: Session, secretbox: SecretBox, manager: MountManager
+) -> None:
+    from netmover.credentials import create_password, list_passwords, update_password
+
+    cred = create_password(session, secretbox, "NAS-Scanner", "scanner", "Pa$$,wort", "WG")
+    a, err = save_mount(session, secretbox, manager, _cifs(cred_id=str(cred.id)))
+    assert not err and a is not None
+    b, err = save_mount(
+        session, secretbox, manager, _cifs(name="archiv", share="archiv", cred_id=str(cred.id))
+    )
+    assert not err and b is not None
+    assert a.credential is cred and b.credential is cred and a.username is None
+    assert list_passwords(session)[0].used_by == ["archiv", "nas"]
+
+    # ein neues Passwort gilt für beide
+    update_password(session, secretbox, cred.id, "scanner", "neu!", "WG")
+    from netmover.admin import credential_for_test, mount_input_from_model
+
+    for m in (a, b):
+        data = mount_input_from_model(m)
+        assert data.cred_id == str(cred.id) and data.username == "scanner"
+        c = credential_for_test(data, m, secretbox, session)
+        assert c is not None and c.secret == "neu!" and c.domain == "WG"
+
+    # Anbindung löschen lässt die gespeicherten Zugangsdaten stehen
+    delete_mount(session, manager, a.id)
+    from netmover.credentials import delete_credential
+
+    with pytest.raises(AdminError, match="noch benutzt"):
+        delete_credential(session, cred.id)
+    delete_mount(session, manager, b.id)
+    assert delete_credential(session, cred.id) == "NAS-Scanner"
+
+
+def test_switch_from_own_to_stored_password_cleans_up(
+    session: Session, secretbox: SecretBox, manager: MountManager
+) -> None:
+    from netmover.credentials import create_password
+
+    m, _ = save_mount(session, secretbox, manager, _cifs(username="alt", password="pw"))
+    assert m is not None and m.credential is not None
+    own_id = m.credential.id
+    cred = create_password(session, secretbox, "geteilt", "scanner", "x")
+    m, err = save_mount(session, secretbox, manager, _cifs(cred_id=str(cred.id)), m.id)
+    assert not err and m is not None and m.credential is cred
+    assert session.get(Credential, own_id) is None
+    # zurück auf eigene Angaben: gespeicherter Eintrag bleibt unverändert
+    m, err = save_mount(session, secretbox, manager, _cifs(username="eigen", password="pw2"), m.id)
+    assert not err and m is not None and not m.credential.managed  # type: ignore[union-attr]
+    assert cred.username == "scanner" and secretbox.decrypt(cred.secret_enc) == "x"
+
+
+def test_stored_password_validation(session: Session, secretbox: SecretBox) -> None:
+    from netmover.credentials import create_password, update_password
+
+    with pytest.raises(AdminError, match="Passwort fehlt"):
+        create_password(session, secretbox, "a", "u", "")
+    with pytest.raises(AdminError):
+        create_password(session, secretbox, "a", "u", "zeile\numbruch")
+    cred = create_password(session, secretbox, "a", "u", "pw")
+    update_password(session, secretbox, cred.id, "u2", "")  # leer = unverändert
+    assert cred.username == "u2" and secretbox.decrypt(cred.secret_enc) == "pw"
+
+
+def test_stored_password_mount_actually_mounts(
+    session_factory: sessionmaker[Session],
+    secretbox: SecretBox,
+    manager: MountManager,
+    fake_backend: object,
+) -> None:
+    from netmover.credentials import create_password
+
+    fake_backend.export("10.0.10.6", "scans")  # type: ignore[attr-defined]
+    with session_scope(session_factory) as s:
+        cred = create_password(s, secretbox, "NAS", "scanner", "Pa$$wort", "WG")
+        m, _ = save_mount(s, secretbox, manager, _cifs(cred_id=str(cred.id)))
+        assert m is not None
+        mid = m.id
+    manager.mount(mid)
+    spec = fake_backend.calls[-1].spec  # type: ignore[attr-defined]
+    assert spec.credentials.username == "scanner" and spec.credentials.password == "Pa$$wort"
+    assert spec.credentials.domain == "WG"
@@ -251,7 +251,7 @@ def test_source_not_deletable_keeps_target_and_prevents_duplicate(
            raise PermissionError(errno.EACCES, "Permission denied", str(path))
        real_unlink(path)  # type: ignore[arg-type]

-    monkeypatch.setattr("netmover.mover.engine.os.unlink", deny)
+    monkeypatch.setattr("netmover.mover.sources.os.unlink", deny)
    first = run(engine, snap(rule_id), src, dst)
    assert first.items[0].status is ItemStatus.FAILED
    assert "nicht löschbar" in (first.items[0].message or "")
@@ -0,0 +1,398 @@
+"""0.5.0: SFTP als Quelle – dieselben Garantien wie bei eingehängten Freigaben."""
+
+from __future__ import annotations
+
+import errno
+import hashlib
+import os
+from collections.abc import Callable, Iterator
+from pathlib import Path
+
+import pytest
+from sqlalchemy import select
+from sqlalchemy.orm import Session, sessionmaker
+
+from netmover.crypto import SecretBox
+from netmover.db import session_scope
+from netmover.models import (
+    CopiedFile,
+    Credential,
+    CredentialKind,
+    ItemStatus,
+    Mount,
+    MountType,
+    PatternType,
+    Rule,
+    RuleAction,
+    RunStatus,
+)
+from netmover.mounting.manager import MountManager
+from netmover.mounting.sftp import SftpConfig, connect
+from netmover.mover.destinations import LocalDestination, SftpDestination
+from netmover.mover.engine import Engine, RuleSnapshot, RunResult
+from netmover.mover.ledger import Ledger
+from netmover.mover.runner import RuleRunner
+from netmover.mover.scan import Matcher
+from netmover.mover.sources import SftpSource
+from netmover.timeouts import CancelToken
+from tests.conftest import MountFactory
+from tests.sftp_stub import StubSFTPServer
+
+
+@pytest.fixture
+def server(tmp_path: Path) -> Iterator[StubSFTPServer]:
+    root = tmp_path / "sftp-root"
+    for d in ("eingang", "ziel", "consume"):
+        (root / d).mkdir(parents=True)
+    srv = StubSFTPServer(root).start()
+    yield srv
+    srv.stop()
+
+
+def _cfg(server: StubSFTPServer) -> SftpConfig:
+    return SftpConfig(
+        "127.0.0.1",
+        server.port,
+        server.username,
+        "/",
+        private_key=server.client_private_key,
+        host_key=server.host_key_string,
+    )
+
+
+@pytest.fixture
+def source(server: StubSFTPServer) -> Iterator[SftpSource]:
+    s = SftpSource(connect(_cfg(server)), "/", "eingang", label="nas2")
+    yield s
+    s.close()
+
+
+@pytest.fixture
+def rule_id(make_mount: MountFactory, session_factory: sessionmaker[Session]) -> int:
+    mid = make_mount()
+    with session_scope(session_factory) as s:
+        r = Rule(name="r", src_mount_id=mid, dst_path="/x")
+        s.add(r)
+        s.flush()
+        return r.id
+
+
+@pytest.fixture
+def hooks() -> list[Callable[[], None]]:
+    return []
+
+
+@pytest.fixture
+def engine(session_factory: sessionmaker[Session], hooks: list[Callable[[], None]]) -> Engine:
+    def sleep(_: float) -> None:
+        for h in hooks:
+            h()
+
+    return Engine(Ledger(session_factory), sleep=sleep)
+
+
+def snap(rule_id: int, **kw: object) -> RuleSnapshot:
+    return RuleSnapshot(id=rule_id, name="r", stability_seconds=2, **kw)  # type: ignore[arg-type]
+
+
+def write(path: Path, data: bytes = b"inhalt") -> Path:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_bytes(data)
+    return path
+
+
+def run(
+    engine: Engine, rule: RuleSnapshot, src: SftpSource, dest: object, **kw: object
+) -> RunResult:
+    return engine.execute(rule, src, dest, CancelToken(), **kw)  # type: ignore[arg-type]
+
+
+def sha(p: Path) -> str:
+    return hashlib.sha256(p.read_bytes()).hexdigest()
+
+
+# -- Scan ----------------------------------------------------------------------------------
+
+
+def test_scan_filters_like_local_sources(source: SftpSource, server: StubSFTPServer) -> None:
+    root = server.root / "eingang"
+    for name in ("a.pdf", "b.PDF", "notiz.txt", ".versteckt.pdf", "halb.pdf.part", "x.tmp"):
+        write(root / name)
+    write(root / "max" / "c.pdf")
+    write(root / "@eaDir" / "a.pdf" / "thumb.pdf")
+    write(root / "quarantaene" / "d.pdf")
+    m = Matcher("*.pdf", PatternType.GLOB)
+    assert [c.relpath for c in source.scan(m, recursive=False)] == ["a.pdf", "b.PDF"]
+    rec = [
+        c.relpath
+        for c in source.scan(m, recursive=True, exclude=frozenset({"eingang/quarantaene"}))
+    ]
+    assert rec == ["a.pdf", "b.PDF", "max/c.pdf"]
+    c = source.scan(m, recursive=False)[0]
+    assert c.path is None and c.state.size == len(b"inhalt")
+
+
+def test_missing_source_dir_aborts(
+    engine: Engine, rule_id: int, server: StubSFTPServer, tmp_path: Path
+) -> None:
+    src = SftpSource(connect(_cfg(server)), "/", "gibtsnicht")
+    try:
+        result = run(
+            engine, snap(rule_id), src, LocalDestination(tmp_path / "dst", create_root=True)
+        )
+    finally:
+        src.close()
+    assert result.status is RunStatus.ERROR and "existiert nicht" in (result.message or "")
+
+
+# -- Verschieben / Kopieren ----------------------------------------------------------------
+
+
+def test_move_sftp_to_local(
+    engine: Engine, rule_id: int, source: SftpSource, server: StubSFTPServer, tmp_path: Path
+) -> None:
+    f = write(server.root / "eingang" / "scan001.pdf", os.urandom(300_000))
+    digest = sha(f)
+    dst = tmp_path / "dst"
+    dst.mkdir()
+    write(dst / "scan001.pdf", b"alt")
+    result = run(engine, snap(rule_id), source, LocalDestination(dst))
+    assert result.status is RunStatus.OK, result.message
+    assert [(i.src, i.status) for i in result.items] == [("scan001.pdf", ItemStatus.MOVED)]
+    assert sha(dst / "scan001-1.pdf") == digest and (dst / "scan001.pdf").read_bytes() == b"alt"
+    assert not f.exists()
+    assert sorted(p.name for p in dst.iterdir()) == ["scan001-1.pdf", "scan001.pdf"]
+
+
+def test_move_sftp_to_sftp(
+    engine: Engine, rule_id: int, source: SftpSource, server: StubSFTPServer
+) -> None:
+    f = write(server.root / "eingang" / "a.pdf", os.urandom(50_000))
+    digest = sha(f)
+    dest = SftpDestination(connect(_cfg(server)), "/ziel", label="nas3")
+    try:
+        result = run(engine, snap(rule_id), source, dest)
+    finally:
+        dest.close()
+    assert result.items[0].status is ItemStatus.MOVED and result.items[0].dst == "nas3:/ziel/a.pdf"
+    assert sha(server.root / "ziel" / "a.pdf") == digest and not f.exists()
+    assert list((server.root / "ziel").glob(".netmover-*")) == []
+
+
+def test_copy_mode_remembers_versions(
+    engine: Engine,
+    rule_id: int,
+    source: SftpSource,
+    server: StubSFTPServer,
+    tmp_path: Path,
+    session_factory: sessionmaker[Session],
+) -> None:
+    f = write(server.root / "eingang" / "a.pdf", b"v1")
+    dst = tmp_path / "dst"
+    dst.mkdir()
+    rule = snap(rule_id, action=RuleAction.COPY)
+    for _ in range(3):
+        run(engine, rule, source, LocalDestination(dst))
+    assert sorted(p.name for p in dst.iterdir()) == ["a.pdf"] and f.exists()
+    f.write_bytes(b"version 2")  # andere Größe → neue Version
+    run(engine, rule, source, LocalDestination(dst))
+    assert sorted(p.name for p in dst.iterdir()) == ["a-1.pdf", "a.pdf"]
+    with session_factory() as s:
+        assert len(s.scalars(select(CopiedFile)).all()) == 2
+
+
+def test_mtime_is_carried_over(
+    engine: Engine, rule_id: int, source: SftpSource, server: StubSFTPServer, tmp_path: Path
+) -> None:
+    f = write(server.root / "eingang" / "a.pdf")
+    os.utime(f, (1_600_000_000, 1_600_000_123))
+    dst = tmp_path / "dst"
+    dst.mkdir()
+    run(engine, snap(rule_id, action=RuleAction.COPY), source, LocalDestination(dst))
+    assert int((dst / "a.pdf").stat().st_mtime) == 1_600_000_123
+
+
+# -- Fallstricke ---------------------------------------------------------------------------
+
+
+def test_growing_file_is_not_touched(
+    engine: Engine,
+    rule_id: int,
+    source: SftpSource,
+    server: StubSFTPServer,
+    tmp_path: Path,
+    hooks: list[Callable[[], None]],
+) -> None:
+    growing = write(server.root / "eingang" / "wachsend.pdf", b"x" * 10)
+    hooks.append(lambda: growing.write_bytes(b"x" * 20))
+    dst = tmp_path / "dst"
+    dst.mkdir()
+    result = run(engine, snap(rule_id), source, LocalDestination(dst))
+    assert result.items[0].status is ItemStatus.SKIPPED and growing.exists()
+    assert list(dst.iterdir()) == []
+
+
+def test_source_changed_during_transfer_keeps_source(
+    engine: Engine,
+    rule_id: int,
+    source: SftpSource,
+    server: StubSFTPServer,
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    f = write(server.root / "eingang" / "a.pdf", b"original")
+    real = LocalDestination._apply_meta
+
+    def change(self: LocalDestination, rel: str, meta: object) -> list[str]:
+        f.write_bytes(b"geaendert!!")
+        return real(self, rel, meta)  # type: ignore[arg-type]
+
+    monkeypatch.setattr(LocalDestination, "_apply_meta", change)
+    dst = tmp_path / "dst"
+    dst.mkdir()
+    result = run(engine, snap(rule_id), source, LocalDestination(dst))
+    assert result.items[0].status is ItemStatus.SKIPPED and f.exists()
+    assert list(dst.iterdir()) == []
+
+
+def test_source_not_deletable_prevents_duplicates(
+    engine: Engine,
+    rule_id: int,
+    source: SftpSource,
+    server: StubSFTPServer,
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    write(server.root / "eingang" / "a.pdf")
+
+    def deny(*a: object, **kw: object) -> None:
+        raise PermissionError(errno.EACCES, "Permission denied")
+
+    monkeypatch.setattr(source.client, "remove", deny)
+    dst = tmp_path / "dst"
+    dst.mkdir()
+    first = run(engine, snap(rule_id), source, LocalDestination(dst))
+    assert first.items[0].status is ItemStatus.FAILED and "nicht löschbar" in (
+        first.items[0].message or ""
+    )
+    second = run(engine, snap(rule_id), source, LocalDestination(dst))
+    assert second.items == [] and sorted(p.name for p in dst.iterdir()) == ["a.pdf"]
+
+
+def test_quarantine_on_sftp_source(
+    engine: Engine,
+    rule_id: int,
+    source: SftpSource,
+    server: StubSFTPServer,
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    from netmover.mover.destinations import TransferError
+
+    def broken(*a: object, **kw: object) -> None:
+        raise TransferError("kaputt")
+
+    monkeypatch.setattr(LocalDestination, "_link_noreplace", broken)
+    write(server.root / "eingang" / "defekt.pdf")
+    dst = tmp_path / "dst"
+    dst.mkdir()
+    rule = snap(rule_id, quarantine_after=2)
+    for _ in range(2):
+        run(engine, rule, source, LocalDestination(dst), quarantine_rel="eingang/_q")
+    r = run(engine, rule, source, LocalDestination(dst), quarantine_rel="eingang/_q")
+    assert r.items[0].status is ItemStatus.QUARANTINED
+    assert r.items[0].dst == "nas2:/eingang/_q/defekt.pdf"
+    assert (server.root / "eingang" / "_q" / "defekt.pdf").exists()
+
+
+def test_connection_loss_aborts_run(
+    engine: Engine,
+    rule_id: int,
+    source: SftpSource,
+    server: StubSFTPServer,
+    tmp_path: Path,
+    hooks: list[Callable[[], None]],
+) -> None:
+    write(server.root / "eingang" / "a.pdf")
+    write(server.root / "eingang" / "b.pdf")
+    hooks.append(lambda: source.session.transport.close())  # Verbindung bricht weg
+    dst = tmp_path / "dst"
+    dst.mkdir()
+    result = run(engine, snap(rule_id), source, LocalDestination(dst))
+    assert result.status is RunStatus.ERROR
+    assert (server.root / "eingang" / "a.pdf").exists() and list(dst.iterdir()) == []
+
+
+# -- Regellauf mit SFTP-Quelle ------------------------------------------------------------
+
+
+def test_runner_with_sftp_source(
+    server: StubSFTPServer,
+    manager: MountManager,
+    session_factory: sessionmaker[Session],
+    secretbox: SecretBox,
+    tmp_path: Path,
+) -> None:
+    write(server.root / "eingang" / "scan.pdf", b"%PDF")
+    consume = tmp_path / "consume"
+    consume.mkdir()
+    with session_scope(session_factory) as s:
+        c = Credential(
+            name="k",
+            kind=CredentialKind.SSH_KEY,
+            username=server.username,
+            secret_enc=secretbox.encrypt(server.client_private_key),
+        )
+        s.add(c)
+        s.flush()
+        m = Mount(
+            name="Scanner-Server",
+            slug="scanner",
+            type=MountType.SFTP,
+            server="127.0.0.1",
+            share="/",
+            port=server.port,
+            cred_id=c.id,
+            host_key=server.host_key_string,
+        )
+        s.add(m)
+        s.flush()
+        r = Rule(
+            name="SFTP → lokal",
+            src_mount_id=m.id,
+            src_subpath="eingang",
+            dst_path=str(consume),
+            stability_seconds=2,
+        )
+        s.add(r)
+        s.flush()
+        rid = r.id
+    result = RuleRunner(session_factory, manager, sleep=lambda _: None).run(rid)
+    assert result.status is RunStatus.OK, result.message
+    assert result.items[0].dst == str(consume / "scan.pdf")
+    assert (consume / "scan.pdf").read_bytes() == b"%PDF"
+    assert not (server.root / "eingang" / "scan.pdf").exists()
+
+
+def test_rule_form_accepts_sftp_source_with_min_stability(
+    session_factory: sessionmaker[Session],
+) -> None:
+    from netmover.admin import parse_rule_form
+
+    with session_scope(session_factory) as s:
+        m = Mount(name="sftp", slug="sftp", type=MountType.SFTP, server="10.0.0.1", share="/in")
+        s.add(m)
+        s.flush()
+        form = {
+            "name": "x",
+            "src_mount_id": str(m.id),
+            "dst_kind": "local",
+            "dst_path": "/tmp",
+            "interval_seconds": "30",
+            "stability_seconds": "1",
+            "op_timeout_seconds": "120",
+        }
+        _, errors = parse_rule_form(s, form)
+        assert "src_mount_id" not in errors and "mindestens 2 s" in errors["stability_seconds"]
+        _, errors = parse_rule_form(s, {**form, "stability_seconds": "3"})
+        assert errors == {}
@@ -512,12 +512,16 @@ def test_export_excludes_secrets_and_import_roundtrip(
 def test_keys_page_generate_import_select_delete(client: TestClient, state: NetMover) -> None:
    from netmover.crypto import generate_ssh_keypair

-    r = post(client, "/keys/generate", {"name": "paperless-lxc"})
+    r = post(client, "/credentials/keys/generate", {"name": "paperless-lxc"})
    assert r.status_code == 303
    private, public = generate_ssh_keypair()
-    r = post(client, "/keys/import", {"name": "alt", "private_key": private, "passphrase": ""})
+    r = post(
+        client,
+        "/credentials/keys/import",
+        {"name": "alt", "private_key": private, "passphrase": ""},
+    )
    assert r.status_code == 303
-    page = client.get("/keys").text
+    page = client.get("/credentials").text
    assert "paperless-lxc" in page and "alt" in page
    assert public.split()[1] in page  # öffentlicher Teil sichtbar
    for line in private.splitlines()[1:-1]:  # keine Zeile des privaten Schlüssels
@@ -546,27 +550,31 @@ def test_keys_page_generate_import_select_delete(client: TestClient, state: NetM
    )
    assert r.status_code == 303, r.text
    mounts = client.get("/mounts").text
-    assert "gespeichertem Schlüssel" in mounts and "„alt“" in mounts
+    assert "gespeicherten Zugangsdaten" in mounts and "„alt“" in mounts
    edit = client.get(f"/mounts/{_mount_id(state, 'paperless')}/edit").text
    assert f'<option value="{key_id}" selected>' in edit and 'value="root"' in edit

-    post(client, f"/keys/{key_id}/delete")
-    assert "noch benutzt" in client.get("/keys").text
+    post(client, f"/credentials/{key_id}/delete")
+    assert "noch benutzt" in client.get("/credentials").text
    post(client, f"/mounts/{_mount_id(state, 'paperless')}/delete")
-    post(client, f"/keys/{key_id}/delete")
-    page = client.get("/keys").text
+    post(client, f"/credentials/{key_id}/delete")
+    page = client.get("/credentials").text
    assert "gelöscht" in page and ">alt<" not in page


 def test_key_import_error_is_shown(client: TestClient) -> None:
-    post(client, "/keys/import", {"name": "x", "private_key": "kein schlüssel", "passphrase": ""})
-    assert "Kein lesbarer privater Schlüssel" in client.get("/keys").text
+    post(
+        client,
+        "/credentials/keys/import",
+        {"name": "x", "private_key": "kein schlüssel", "passphrase": ""},
+    )
+    assert "Kein lesbarer privater Schlüssel" in client.get("/credentials").text


 def test_no_escaped_attributes_in_pages(client: TestClient, state: NetMover) -> None:
    """Attribute in {{ … }} werden von Jinja maskiert – dann landet &#34; im HTML."""

-    post(client, "/keys/generate", {"name": "k"})
+    post(client, "/credentials/keys/generate", {"name": "k"})
    for url in (
        "/",
        "/mounts",
@@ -581,3 +589,35 @@ def test_no_escaped_attributes_in_pages(client: TestClient, state: NetMover) ->
        html = client.get(url).text
        broken = re.findall(r"<[a-z][^<>]*=&#34;[^<>]*>", html)  # nur in Tags, nicht im Text
        assert not broken, (url, broken[:2])
+
+
+def test_credentials_page_passwords(
+    client: TestClient, state: NetMover, backend: FakeMountBackend
+) -> None:
+    r = post(
+        client,
+        "/credentials/passwords",
+        {"name": "NAS-Scanner", "username": "scanner", "password": "Geheim,123", "domain": ""},
+    )
+    assert r.status_code == 303
+    page = client.get("/credentials").text
+    assert "NAS-Scanner" in page and "scanner" in page and "Geheim,123" not in page
+    with state.session_factory() as s:
+        cid = s.scalar(select(Credential.id).where(Credential.name == "NAS-Scanner"))
+    form = client.get("/mounts/new").text
+    assert f'<option value="{cid}"' in form and "eigene Angaben für diese Anbindung" in form
+    backend.export("nas", "scans")
+    r = post(
+        client, "/mounts/new", {**MOUNT_FORM, "username": "", "password": "", "cred_id": str(cid)}
+    )
+    assert r.status_code == 303, r.text
+    assert "gespeicherten Zugangsdaten" in client.get("/mounts").text
+    mid = _mount_id(state, "Synology Scans")
+    assert "Quelle mit Löschen" in post(client, f"/mounts/{mid}/test").text
+    post(
+        client,
+        f"/credentials/{cid}/password",
+        {"username": "scanner2", "domain": "", "password": ""},
+    )
+    assert "gilt ab sofort für alle" in client.get("/credentials").text
+    assert client.get("/keys").headers["location"] == "/credentials"
Author	SHA1	Message	Date
Florian WildeandClaude Opus 5.5	579a2c164f	feat: Anmeldung abschaltbar (vertraute Netze oder ganz); Doku 0.5.0 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>	2026-10-11 01:12:28 +02:00
Florian WildeandClaude Opus 5.5	f5ea5840d4	feat: gespeicherte Zugangsdaten für SMB und SFTP-Passwörter Seite „Zugangsdaten“ (vorher „Schlüssel“) mit Passwörtern und SSH-Schlüsseln; Anbindungen wählen gespeicherte Passwörter aus, ein neues Passwort gilt sofort für alle. /keys leitet auf /credentials um. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>	2026-10-11 01:07:09 +02:00
Florian WildeandClaude Opus 5.5	893fe0c95d	feat(web): SFTP-Anbindungen als Quelle wählbar; Doku Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>	2026-10-11 01:00:30 +02:00
Florian WildeandClaude Opus 5.5	5786b1e206	feat: SFTP als Quelle mit denselben Garantien wie SMB/NFS Quell-Abstraktion (LocalSource, SftpSource): Scan, frische Zustände, Lesen, Löschen und Quarantäne über eine Schnittstelle. Verbindungsabbrüche brechen den Lauf als gestörte Anbindung ab. Bei SFTP-Quellen mindestens 2 s Stabilitätsfenster (Zeiten nur sekundengenau). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>	2026-10-11 01:00:30 +02:00