feat: Anbindungen kopieren – inklusive Zugangsdaten

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

CHANGELOG.md

@@ -8,6 +8,41 @@ Die Version steht an genau einer Stelle: `src/netmover/__init__.py`.
 
 ## [Unreleased]
 
+## [0.8.0] – 2026-10-12
+
+### Hinzugefügt
+- **Anbindungen kopieren** (Knopf „Kopieren“): Formular vorausgefüllt, Name
+  „… (Kopie)“. Eigene Zugangsdaten der Vorlage werden übernommen (eigenes Exemplar),
+  solange kein neues Passwort eingegeben wird; ein eigener SSH-Schlüssel wird
+  weiterverwendet; gespeicherte Zugangsdaten bleiben ausgewählt. „Testen“ im
+  Kopie-Formular nutzt die übernommenen Zugangsdaten. Der SFTP-Hostschlüssel wird nur
+  bei gleichem Server und Port übernommen.
+
+## [0.7.0] – 2026-10-12
+
+### Hinzugefügt
+- **Archiv je Regel:** zusätzliche Kopie in ein Archiv (Anbindung SMB/NFS/SFTP oder
+  lokaler Pfad), Ablage „wie in der Quelle“ oder „nach Datum“ (`Jahr/Monat/…`).
+  Reihenfolge: erst Archiv, dann Ziel, **erst dann wird die Quelle gelöscht**.
+  Scheitert das Ziel nach dem Archiv, verhindert eine Buchführung
+  (`archived_files`) eine zweite Archivkopie. Im Archiv wird nie überschrieben.
+  Nicht erreichbares Archiv → Regel wartet. Platzprüfung auch fürs Archiv,
+  Probelauf zeigt das Archivziel. Migration 0004.
+- Export/Import enthalten die Archiv-Einstellungen; eine als Archiv benutzte
+  Anbindung lässt sich nicht löschen.
+
+## [0.6.0] – 2026-10-11
+
+### Geändert
+- **Anmeldung je Netz konfigurierbar** (*Einstellungen → Anmeldung*): „Ohne Passwort
+  aus diesen Netzen“ und „Mit Passwort aus diesen Netzen“ (leer = alle übrigen).
+  Ist die zweite Liste ausgefüllt, werden alle anderen Netze abgewiesen (403, auch die
+  Anmeldeseite); `/healthz` und statische Dateien bleiben erreichbar. Ersetzt die
+  Auswahl „immer / vertraute Netze / nie“ aus 0.5.0 (`0.0.0.0/0, ::/0` = ganz aus;
+  gespeicherter Modus „nie“ gilt weiter).
+- Schutz vor Aussperren: Eine Einstellung, mit der die eigene Adresse keinen Zugang
+  mehr hätte, wird nicht gespeichert. `netmover reset-password` hebt auch Netz-Sperren auf.
+
 ## [0.5.0] – 2026-10-11
 
 ### Hinzugefügt
@@ -234,7 +269,10 @@ 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.5.0...HEAD
+[Unreleased]: https://git.wilde-server.de/florian.wilde/netmover/compare/v0.8.0...HEAD
+[0.8.0]: https://git.wilde-server.de/florian.wilde/netmover/compare/v0.7.0...v0.8.0
+[0.7.0]: https://git.wilde-server.de/florian.wilde/netmover/compare/v0.6.0...v0.7.0
+[0.6.0]: https://git.wilde-server.de/florian.wilde/netmover/compare/v0.5.0...v0.6.0
 [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

DEVLOG.md

@@ -19,10 +19,63 @@ Zeiten in Ortszeit (Europe/Berlin).
 | 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) |
 | 2026-10-11 | 00:54 | 01:16 | 0:22  | –       | SFTP-Quelle, Zugangsdaten, Anmeldung (0.5.0) |
-| **Summe**  |       |       | **3:17** |      |                                        |
+| 2026-10-11 | 08:43 | 08:55 | 0:12  | –       | Anmeldung je Netz (0.6.0)              |
+| 2026-10-12 | 00:14 | 00:34 | 0:20  | –       | Archiv je Regel (0.7.0)                |
+| 2026-10-12 | 00:46 | 00:55 | 0:09  | –       | Anbindungen kopieren (0.8.0)           |
+| **Summe**  |       |       | **3:58** |      |                                        |
 
 ---
 
+## 2026-10-12 · 00:46–00:55 · Anbindungen kopieren (0.8.0)
+
+**Erledigt** – Knopf „Kopieren“, Route `/mounts/{id}/copy` (GET Formular, POST
+speichern), `save_mount(copy_from=…)`, Formular-Test mit Vorlage. 7 Tests (357 gesamt);
+Gegenprobe ohne Übernahme der Zugangsdaten: 3 Tests schlagen an.
+
+**Entscheidungen**
+- Eigene Zugangsdaten werden als eigenes Exemplar kopiert (nicht geteilt) – eigene
+  Zugangsdaten verschwinden mit ihrer Anbindung, eine geteilte Nutzung wäre
+  dafür der Fall „Zugangsdaten“.
+- Eigenes Schlüsselpaar bleibt gleich (schon in `authorized_keys`).
+- Hostschlüssel nur bei gleichem Server/Port übernehmen, sonst würde der Test an
+  einem anderen Server mit „Hostschlüssel geändert“ scheitern.
+
+## 2026-10-12 · 00:14–00:34 · Archiv je Regel (0.7.0)
+
+**Erledigt**
+- Archivfelder an der Regel, Tabelle `archived_files` (Migration 0004).
+- Engine: Archiv zuerst, dann Ziel, dann Quelle löschen; Archiv immer „umbenennen“.
+- Runner: Ziel- und Archivaufbau über eine gemeinsame Fabrik (`_target`).
+- 14 Tests (350 gesamt); Mutationsprobe 4/4 erkannt (Buchführung, Archiv
+  übersprungen, Kollisionsregel, Platzprüfung).
+
+**Entscheidungen**
+- Archiv *vor* Ziel: Paperless holt Dateien sofort ab – andersherum würde ein
+  gescheitertes Archiv beim Neuversuch zur doppelten Lieferung führen.
+- Archiv überschreibt nie (unabhängig von der Kollisionsregel des Ziels).
+- Ablage „nach Datum“ nach Verarbeitungsdatum (nicht Datei-mtime) – eindeutig
+  nachvollziehbar, wann eine Datei durchgelaufen ist.
+
+**Stolperstein**
+- Downgrade von 0004 scheiterte an den CHECK-Regeln der neuen Auswahlfelder
+  (verweisen nach dem Entfernen der Spalten ins Leere) – nur dank der Migrations-
+  tests mit Daten aufgefallen; Downgrade entfernt sie jetzt zuerst.
+
+## 2026-10-11 · 08:43–08:55 · Anmeldung je Netz (0.6.0)
+
+**Anlass** – Wunsch: Anmeldung abschalten, und zwar konfigurierbar, aus welchen Netzen
+mit und ohne Passwort. Zugriffe kamen aus 10.0.10.0/24 und 10.1.5.0/24 (Access-Log).
+
+**Erledigt**
+- `auth.access_decision()` als einzige Entscheidung (ohne PW / mit PW / abgewiesen),
+  benutzt von Middleware (403 für fremde Netze), Guard und Anmeldeseite.
+- Schutz vor Aussperren beim Speichern; `reset-password` hebt Sperren auf.
+- 19 Tests für die Anmeldung (336 gesamt).
+
+**Entscheidungen**
+- `/healthz` und `/static` nie sperren – Uptime Kuma kann in einem anderen Netz stehen.
+- Nicht-IP-Clients (z. B. Unix-Socket) gelten als „in keinem Netz“.
+
 ## 2026-10-11 · 00:54–01:16 · SFTP-Quelle, Zugangsdaten, Anmeldung abschaltbar (0.5.0)
 
 **Erledigt**

ROADMAP.md

@@ -28,6 +28,9 @@ Legende: ✅ erledigt · 🔨 in Arbeit · 📋 geplant · 💡 Idee (noch nicht
 
 | Version | Thema |
 |---|---|
+| 0.8.0 | Anbindungen kopieren inklusive Zugangsdaten |
+| 0.7.0 | Archiv je Regel: Kopie ins Ziel und ins Archiv, Quelle erst danach löschen |
+| 0.6.0 | Anmeldung je Netz: ohne Passwort / mit Passwort / abgewiesen, Schutz vor Aussperren |
 | 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) |

compose.yml

@@ -4,7 +4,7 @@
 services:
   netmover:
     build: .
-    image: netmover:0.5.0
+    image: netmover:0.8.0
     container_name: netmover
     restart: unless-stopped
     cap_add:

docs/ANLEITUNG.md

@@ -8,7 +8,7 @@ Diese Anleitung erklärt die Bedienung. Installation, Updates und technische Det
 stehen im [README](../README.md).
 
 **Inhalt:** [Begriffe](#begriffe) · [Erste Anmeldung](#erste-anmeldung) ·
-[Zugangsdaten](#zugangsdaten) · [Anbindungen](#anbindungen) · [Regeln](#regeln) ·
+[Zugangsdaten](#zugangsdaten) · [Anbindungen](#anbindungen) · [Regeln](#regeln) · [Archiv](#archiv) ·
 [Probelauf](#probelauf) · [Status](#status) · [Wartungsmodus](#wartungsmodus) · [Protokoll](#protokoll) ·
 [Einstellungen](#einstellungen) · [Anmeldung](#anmeldung) · [Uptime Kuma](#uptime-kuma) · [Beispiele](#beispiele) · [Häufige Fragen](#fragen)
 
@@ -72,6 +72,22 @@ Stabilitätsfenster dort mindestens 2 Sekunden betragen.
 | **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. |
 | **Aktiv** | Deaktivierte Anbindungen werden von Regeln nicht benutzt. |
 
+### Kopieren {#anbindung-kopieren}
+
+Für eine ähnliche Anbindung – etwa eine zweite Freigabe auf demselben NAS – bei der
+Vorlage auf **Kopieren** klicken. Das Formular ist mit allen Werten vorausgefüllt
+(Name „… (Kopie)“); meist muss nur die Freigabe geändert werden.
+
+- **Zugangsdaten werden übernommen**, solange das Passwortfeld leer bleibt. Die Kopie
+  bekommt ein eigenes Exemplar – das Löschen der Vorlage betrifft sie nicht.
+  Gespeicherte Zugangsdaten sind einfach vorausgewählt.
+- Ein **eigener SSH-Schlüssel** der Vorlage wird weiter benutzt; ein neuer Eintrag in
+  `authorized_keys` ist nicht nötig.
+- **Testen** funktioniert im Kopie-Formular bereits mit den übernommenen Zugangsdaten.
+
+Tipp: Nutzen mehrere Anbindungen dieselben Zugangsdaten, lohnt sich ein Eintrag unter
+[Zugangsdaten](#zugangsdaten) – dann gilt ein neues Passwort sofort überall.
+
 ### Testen {#anbindung-testen}
 
 **Testen** prüft drei Dinge einzeln und sagt im Klartext, was nicht klappt:
@@ -160,6 +176,34 @@ Gespeicherte Zugangsdaten lassen sich nur löschen, wenn sie keine Anbindung meh
 | **Aktion** | **verschieben** (Quelle wird nach erfolgreicher Übertragung gelöscht) oder **kopieren** (Quelle bleibt; jede Dateiversion wird nur einmal kopiert). |
 | **Wenn Datei existiert** | **umbenennen** (`scan.pdf` → `scan-1.pdf`, `scan-2.pdf` …), **überspringen** oder **überschreiben**. |
 
+### Archiv {#archiv}
+
+Optional bekommt jede Datei zusätzlich eine **Kopie ins Archiv** – z. B. Paperless als
+Ziel und ein Archivordner auf dem NAS. **Die Quelle wird erst gelöscht, wenn die Datei
+im Ziel UND im Archiv angekommen ist.**
+
+| Feld | Bedeutung |
+|---|---|
+| **Archiv** | kein Archiv / eine Anbindung (SMB, NFS, SFTP) / ein lokaler Pfad |
+| **Archivordner** | Anbindung: relativer Unterordner (wird angelegt). Lokal: absoluter Pfad, muss existieren. |
+| **Ablage** | **wie in der Quelle** oder **nach Datum** – `2026/10/scan001.pdf` (Monat der Verarbeitung) |
+
+So läuft es pro Datei:
+
+1. **Zuerst ins Archiv** – mit denselben Sicherungen wie beim Ziel.
+2. **Dann ins Ziel.**
+3. **Dann erst** wird die Quelle gelöscht (bei „verschieben“).
+
+Das Archiv kommt zuerst, weil ein Ziel wie Paperless Dateien sofort abholt: Scheitert
+das Archiv, wurde noch nichts geliefert, und die Datei bleibt in der Quelle. Scheitert
+danach das Ziel, merkt sich NetMover, dass die Datei schon archiviert ist – der nächste
+Versuch legt keine zweite Archivkopie an.
+
+Im Archiv wird **nie überschrieben**: Gleiche Namen werden immer umbenannt
+(`scan-1.pdf`), egal was die Regel fürs Ziel vorsieht. Ist das Archiv nicht erreichbar,
+wartet die Regel wie bei Quelle und Ziel. Im Kopiermodus bekommen Ziel und Archiv je
+eine Kopie, die Quelle bleibt.
+
 ### Sicherheit und Grenzen {#regel-sicherheit}
 
 | Feld | Bedeutung | Empfehlung |
@@ -333,25 +377,33 @@ dreimal hintereinander gescheitert.
 
 ### Anmeldung {#anmeldung}
 
-Standardmäßig fragt NetMover immer nach dem Passwort. Unter *Einstellungen →
+Unter *Einstellungen → Anmeldung* legst du fest, **aus welchen Netzen** es mit und
-Anmeldung* lässt sich das ändern (Bestätigung mit dem aktuellen Passwort):
+ohne Passwort geht (Bestätigung mit dem aktuellen Passwort):
 
-| Einstellung | Wirkung |
+| Feld | Wirkung |
 |---|---|
-| **immer** (empfohlen) | Anmeldung bei jedem Zugriff |
+| **Ohne Passwort aus diesen Netzen** | z. B. `10.0.10.0/24, 10.1.5.0/24` – von dort keine Anmeldung. Leer = niemand. `0.0.0.0/0, ::/0` = Anmeldung ganz aus. |
-| **nicht aus vertrauenswürdigen Netzen** | ohne Passwort aus den eingetragenen Netzen, z. B. `10.0.10.0/24`; von überall sonst mit Passwort |
+| **Mit Passwort aus diesen Netzen** | Leer = **alle übrigen** Netze mit Passwort. Ausgefüllt = nur von dort mit Passwort, **alle anderen Netze werden abgewiesen** (auch die Anmeldeseite). |
-| **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
+Beispiel: Ohne Passwort aus dem Heimnetz `10.0.10.0/24`, mit Passwort aus dem
+VPN-Netz `10.8.0.0/24`, alles andere gesperrt.
+
+- Einzelne Adressen gehen auch (`192.168.1.20`).
+- Die Seite zeigt deine aktuelle Adresse. Eine Einstellung, mit der du dich selbst
+  aussperren würdest, wird nicht gespeichert.
+- `/healthz` bleibt für die Überwachung von überall erreichbar.
+- Solange du ohne Passwort arbeitest, 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
+> Oberfläche ohne Passwort erreicht, kann alles ändern – Netze „ohne Passwort“ daher
-> demselben Server, kommen alle Anfragen von `127.0.0.1` – dann `127.0.0.1` nicht als
+> so eng wie möglich wählen. Läuft ein Reverse Proxy auf demselben Server, kommen
-> vertrauenswürdig eintragen und `NETMOVER_FORWARDED_ALLOW_IPS` setzen.
+> alle Anfragen von `127.0.0.1`: dann `127.0.0.1` nicht eintragen und
+> `NETMOVER_FORWARDED_ALLOW_IPS` setzen.
 
-Zurück zur Anmeldung: dieselbe Einstellung auf „immer“, oder auf dem Server
+Ausgesperrt oder Passwort vergessen? Auf dem Server
-`/opt/netmover/venv/bin/netmover reset-password` (setzt auch die Anmeldung zurück).
+`/opt/netmover/venv/bin/netmover reset-password` – setzt das Passwort zurück, schaltet
+die Anmeldung überall wieder ein und hebt Netz-Sperren auf.
 
 ### Export und Import {#export-import}
 
@@ -370,7 +422,8 @@ deaktiviert, Zugangsdaten müssen neu eingegeben werden.
 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,
-   verschieben, umbenennen, 30 s. **Probelauf**, dann aktivieren.
+   verschieben, umbenennen, 30 s. Optional **Archiv**: Anbindung NAS, Ordner
+   `Archiv/Scans`, Ablage nach Datum. **Probelauf**, dann aktivieren.
 
 ### Jede Person in einen eigenen Ordner {#beispiel-unterordner}
 

src/netmover/__init__.py

@@ -1,3 +1,3 @@
 """NetMover – NFS/SMB-Mounts verwalten und Dateien zeitgesteuert verschieben."""
 
-__version__ = "0.5.0"
+__version__ = "0.8.0"

src/netmover/admin.py

@@ -18,6 +18,8 @@ from sqlalchemy.orm import Session
 
 from netmover.crypto import SecretBox, generate_ssh_keypair
 from netmover.models import (
+    ArchiveKind,
+    ArchiveLayout,
     CollisionPolicy,
     Credential,
     CredentialKind,
@@ -292,7 +294,11 @@ def save_mount(
     mounts: MountManager,
     data: MountInput,
     mount_id: int | None = None,
+    copy_from: Mount | None = None,
 ) -> tuple[Mount | None, Errors]:
+    """``copy_from``: Vorlage beim Kopieren – eigene Zugangsdaten (Passwort bzw.
+    Schlüsselpaar) werden übernommen, solange kein neues Passwort eingegeben ist;
+    die Kopie bekommt ein eigenes Exemplar."""
     errors: Errors = {}
     m = s.get(Mount, mount_id) if mount_id is not None else None
     if mount_id is not None and m is None:
@@ -347,6 +353,14 @@ def save_mount(
         setattr(m, f, getattr(data, f))
     if not data.type.is_kernel_mount:
         m.auto_mount = False
+    if (
+        mount_id is None
+        and copy_from is not None
+        and data.type is MountType.SFTP
+        and copy_from.type is MountType.SFTP
+        and (copy_from.server, copy_from.port) == (data.server, data.port)
+    ):
+        m.host_key = copy_from.host_key  # gleicher Server – Hostschlüssel bleibt gültig
 
     # Zugangsdaten: entweder ein gespeicherter Schlüssel aus der Schlüsselverwaltung
     # (geteilt, wird hier nie verändert) oder eigene Zugangsdaten nur dieser Anbindung.
@@ -384,21 +398,32 @@ def save_mount(
     else:
         m.username = None
         cred = own
+        template = (
+            copy_from.credential
+            if copy_from is not None
+            and copy_from.credential is not None
+            and not copy_from.credential.managed
+            and copy_from.credential.kind is data.auth
+            else None
+        )
         if cred is None or cred.kind is not data.auth:
-            if data.auth is CredentialKind.SSH_KEY:
+            if template is not None and (data.auth is CredentialKind.SSH_KEY or not data.password):
-                private, public = generate_ssh_keypair(f"netmover@{m.slug}")
+                # Kopie: dasselbe Passwort bzw. Schlüsselpaar (steht schon in authorized_keys)
-                secret, pub = private, public
+                secret_enc, pub = template.secret_enc, template.public_key
+            elif data.auth is CredentialKind.SSH_KEY:
+                private, pub = generate_ssh_keypair(f"netmover@{m.slug}")
+                secret_enc = secrets.encrypt(private)
             else:
                 if not data.password:
                     errors["password"] = "Passwort fehlt."  # noqa: S105
                     s.rollback()
                     return None, errors
-                secret, pub = data.password, None
+                secret_enc, pub = secrets.encrypt(data.password), None
             if cred is None:
                 cred = Credential(name=f"mount:{m.slug}", managed=False)
                 s.add(cred)
             cred.kind = data.auth
-            cred.secret_enc = secrets.encrypt(secret)
+            cred.secret_enc = secret_enc
             cred.public_key = pub
         elif data.password and data.auth is CredentialKind.PASSWORD:
             cred.secret_enc = secrets.encrypt(data.password)
@@ -419,12 +444,26 @@ def save_mount(
     return m, errors
 
 
+def copy_name(s: Session, name: str) -> str:
+    """„NAS (Kopie)“, „NAS (Kopie 2)“, … – frei in der Datenbank."""
+    base = f"{name} (Kopie)"[:100]
+    candidate, n = base, 1
+    while s.scalar(select(Mount.id).where(Mount.name == candidate)) is not None:
+        n += 1
+        candidate = f"{name} (Kopie {n})"[:100]
+    return candidate
+
+
 def delete_mount(s: Session, mounts: MountManager, mount_id: int) -> None:
     m = s.get(Mount, mount_id)
     if m is None:
         return
     used = s.scalars(
-        select(Rule.name).where((Rule.src_mount_id == mount_id) | (Rule.dst_mount_id == mount_id))
+        select(Rule.name).where(
+            (Rule.src_mount_id == mount_id)
+            | (Rule.dst_mount_id == mount_id)
+            | (Rule.archive_mount_id == mount_id)
+        )
     ).all()
     if used:
         raise AdminError(f"Wird noch von Regeln benutzt: {', '.join(used)}")
@@ -465,6 +504,10 @@ class RuleInput:
     dry_run: bool = False
     quarantine_path: str = ""
     quarantine_after: int = 3
+    archive_kind: ArchiveKind = ArchiveKind.NONE
+    archive_mount_id: int | None = None
+    archive_path: str = ""
+    archive_layout: ArchiveLayout = ArchiveLayout.FLAT
     extra: dict[str, str] = field(default_factory=dict)
 
 
@@ -492,6 +535,10 @@ def rule_input_from_model(r: Rule) -> RuleInput:
         dry_run=r.dry_run,
         quarantine_path=r.quarantine_path or "",
         quarantine_after=r.quarantine_after,
+        archive_kind=r.archive_kind,
+        archive_mount_id=r.archive_mount_id,
+        archive_path=r.archive_path,
+        archive_layout=r.archive_layout,
     )
 
 
@@ -585,6 +632,39 @@ def parse_rule_form(s: Session, form: Mapping[str, str]) -> tuple[RuleInput, Err
             d.quarantine_path = safe_relpath("quarantine_path", d.quarantine_path)
         except ValidationError as exc:
             errors["quarantine_path"] = str(exc)
+
+    # Archiv
+    try:
+        d.archive_kind = ArchiveKind(form.get("archive_kind") or "none")
+        d.archive_layout = ArchiveLayout(form.get("archive_layout") or "flat")
+    except ValueError:
+        errors["archive_kind"] = "Ungültige Auswahl."
+    d.archive_path = (form.get("archive_path") or "").strip()
+    if d.archive_kind is ArchiveKind.MOUNT:
+        d.archive_mount_id = _int(form, "archive_mount_id", errors)
+        if d.archive_mount_id is None or s.get(Mount, d.archive_mount_id) is None:
+            errors["archive_mount_id"] = "Archiv-Anbindung wählen."
+        try:
+            d.archive_path = safe_relpath("archive_path", d.archive_path)
+        except ValidationError as exc:
+            errors["archive_path"] = str(exc)
+        if (
+            "archive_path" not in errors
+            and d.archive_mount_id == d.dst_mount_id
+            and d.dst_kind == "mount"
+            and d.archive_path == d.dst_path
+        ):
+            errors["archive_path"] = "Archiv und Ziel sind derselbe Ordner."
+    elif d.archive_kind is ArchiveKind.LOCAL:
+        d.archive_mount_id = None
+        try:
+            d.archive_path = safe_abspath("archive_path", d.archive_path)
+        except ValidationError as exc:
+            errors["archive_path"] = str(exc)
+        if d.dst_kind == "local" and d.archive_path == d.dst_path:
+            errors["archive_path"] = "Archiv und Ziel sind derselbe Ordner."
+    else:
+        d.archive_mount_id, d.archive_path = None, ""
     return d, errors
 
 
@@ -617,6 +697,10 @@ def save_rule(s: Session, d: RuleInput, rule_id: int | None = None) -> tuple[Rul
         "on_collision",
         "dry_run",
         "quarantine_after",
+        "archive_kind",
+        "archive_mount_id",
+        "archive_path",
+        "archive_layout",
     ):
         setattr(r, f, getattr(d, f))
     r.set_owner = d.set_owner or None

src/netmover/cli.py

@@ -64,10 +64,15 @@ 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)
+            # Anmeldung wieder überall einschalten und Netz-Sperren aufheben
-            mode = s.get(Setting, "app.auth_mode")
+            for key, value in (
-            if mode is not None:
+                ("app.auth_mode", '"always"'),
-                mode.value = '"always"'
+                ("app.trusted_networks", '""'),
+                ("app.allowed_networks", '""'),
+            ):
+                setting = s.get(Setting, key)
+                if setting is not None:
+                    setting.value = value
             s.commit()
         engine.dispose()
         print("Passwort gelöscht. Dienst neu starten; der Einrichtungscode steht im Log.")

src/netmover/migrations/versions/0004_archive.py

@@ -0,0 +1,103 @@
+"""archive
+
+Revision ID: 0004
+Revises: 0003
+Create Date: 2026-10-12
+"""
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+
+import netmover.models
+
+revision: str = "0004"
+down_revision: str | None = "0003"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "archived_files",
+        sa.Column("id", sa.Integer(), nullable=False),
+        sa.Column("rule_id", sa.Integer(), nullable=False),
+        sa.Column("src_relpath", sa.String(length=2048), nullable=False),
+        sa.Column("size", sa.BigInteger(), nullable=False),
+        sa.Column("mtime_ns", sa.BigInteger(), nullable=False),
+        sa.Column("archive_path", sa.String(length=2048), nullable=False),
+        sa.Column("archived_at", netmover.models.UTCDateTime(), nullable=False),
+        sa.ForeignKeyConstraint(
+            ["rule_id"],
+            ["rules.id"],
+            name=op.f("fk_archived_files_rule_id_rules"),
+            ondelete="CASCADE",
+        ),
+        sa.PrimaryKeyConstraint("id", name=op.f("pk_archived_files")),
+        sa.UniqueConstraint(
+            "rule_id",
+            "src_relpath",
+            "size",
+            "mtime_ns",
+            name=op.f("uq_archived_files_rule_id_src_relpath_size_mtime_ns"),
+        ),
+    )
+    with op.batch_alter_table("rules", schema=None) as batch_op:
+        batch_op.add_column(
+            sa.Column(
+                "archive_kind",
+                sa.Enum(
+                    "none",
+                    "mount",
+                    "local",
+                    name="archive_kind",
+                    native_enum=False,
+                    create_constraint=True,
+                    length=24,
+                ),
+                server_default="none",
+                nullable=False,
+            )
+        )
+        batch_op.add_column(sa.Column("archive_mount_id", sa.Integer(), nullable=True))
+        batch_op.add_column(
+            sa.Column("archive_path", sa.String(length=1024), server_default="", nullable=False)
+        )
+        batch_op.add_column(
+            sa.Column(
+                "archive_layout",
+                sa.Enum(
+                    "flat",
+                    "date",
+                    name="archive_layout",
+                    native_enum=False,
+                    create_constraint=True,
+                    length=24,
+                ),
+                server_default="flat",
+                nullable=False,
+            )
+        )
+        batch_op.create_foreign_key(
+            batch_op.f("fk_rules_archive_mount_id_mounts"),
+            "mounts",
+            ["archive_mount_id"],
+            ["id"],
+            ondelete="RESTRICT",
+        )
+
+
+def downgrade() -> None:
+    with op.batch_alter_table("rules", schema=None) as batch_op:
+        batch_op.drop_constraint(batch_op.f("fk_rules_archive_mount_id_mounts"), type_="foreignkey")
+        # CHECK-Regeln der Auswahlfelder zuerst entfernen – sonst verweisen sie beim
+        # Neuaufbau der Tabelle auf Spalten, die es nicht mehr gibt.
+        batch_op.drop_constraint(batch_op.f("ck_rules_archive_kind"), type_="check")
+        batch_op.drop_constraint(batch_op.f("ck_rules_archive_layout"), type_="check")
+        batch_op.drop_column("archive_layout")
+        batch_op.drop_column("archive_path")
+        batch_op.drop_column("archive_mount_id")
+        batch_op.drop_column("archive_kind")
+
+    op.drop_table("archived_files")

src/netmover/models.py

@@ -10,6 +10,7 @@ Abweichungen bzw. Ergänzungen gegenüber dem Mindestumfang aus der Spezifikatio
 * ``credentials.managed`` + ``mounts.username`` (Migration 0002) –
   Schlüsselverwaltung: gespeicherte SSH-Schlüssel, die mehrere Anbindungen mit
   jeweils eigenem Benutzernamen nutzen.
+* ``rules.archive_*`` + Tabelle ``archived_files`` (Migration 0004) – Archivkopie.
 * ``rules.waiting_since`` / ``rules.waiting_reason`` (Migration 0003) – Regel wartet
   auf eine nicht erreichbare Anbindung (automatisch pausiert).
 * ``mounts.state_since`` – für „hängt seit …“ / „gestört seit …“ auf der Statusseite.
@@ -138,6 +139,17 @@ class CollisionPolicy(enum.StrEnum):
     OVERWRITE = "overwrite"
 
 
+class ArchiveKind(enum.StrEnum):
+    NONE = "none"
+    MOUNT = "mount"
+    LOCAL = "local"
+
+
+class ArchiveLayout(enum.StrEnum):
+    FLAT = "flat"  # wie in der Quelle
+    DATE = "date"  # Jahr/Monat/…, nach Tag der Verarbeitung
+
+
 class RunStatus(enum.StrEnum):
     RUNNING = "running"
     OK = "ok"  # keine Dateifehler
@@ -259,6 +271,18 @@ class Rule(Base):
     quarantine_path: Mapped[str | None] = mapped_column(String(1024))
     quarantine_after: Mapped[int] = mapped_column(Integer, default=3)
 
+    # Archiv (Migration 0004): zusätzliche Kopie; Quelle erst löschen, wenn Ziel UND Archiv da
+    archive_kind: Mapped[ArchiveKind] = mapped_column(
+        _enum(ArchiveKind, "archive_kind"), default=ArchiveKind.NONE, server_default="none"
+    )
+    archive_mount_id: Mapped[int | None] = mapped_column(
+        ForeignKey("mounts.id", ondelete="RESTRICT")
+    )
+    archive_path: Mapped[str] = mapped_column(String(1024), default="", server_default="")
+    archive_layout: Mapped[ArchiveLayout] = mapped_column(
+        _enum(ArchiveLayout, "archive_layout"), default=ArchiveLayout.FLAT, server_default="flat"
+    )
+
     last_run_at: Mapped[datetime | None] = mapped_column(UTCDateTime)
     last_success_at: Mapped[datetime | None] = mapped_column(UTCDateTime)
     consecutive_failures: Mapped[int] = mapped_column(Integer, default=0)
@@ -269,6 +293,7 @@ class Rule(Base):
 
     src_mount: Mapped[Mount] = relationship(foreign_keys=[src_mount_id])
     dst_mount: Mapped[Mount | None] = relationship(foreign_keys=[dst_mount_id])
+    archive_mount: Mapped[Mount | None] = relationship(foreign_keys=[archive_mount_id])
     runs: Mapped[list[Run]] = relationship(
         back_populates="rule", cascade="all, delete-orphan", passive_deletes=True
     )
@@ -330,6 +355,22 @@ class CopiedFile(Base):
     copied_at: Mapped[datetime] = mapped_column(UTCDateTime, default=utcnow)
 
 
+class ArchivedFile(Base):
+    """Welche Dateiversion schon im Archiv liegt – damit ein Neuversuch nach einem
+    gescheiterten Ziel keine zweite Archivkopie anlegt."""
+
+    __tablename__ = "archived_files"
+    __table_args__ = (UniqueConstraint("rule_id", "src_relpath", "size", "mtime_ns"),)
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True)
+    rule_id: Mapped[int] = mapped_column(ForeignKey("rules.id", ondelete="CASCADE"))
+    src_relpath: Mapped[str] = mapped_column(String(2048))
+    size: Mapped[int] = mapped_column(BigInteger)
+    mtime_ns: Mapped[int] = mapped_column(BigInteger)
+    archive_path: Mapped[str] = mapped_column(String(2048))
+    archived_at: Mapped[datetime] = mapped_column(UTCDateTime, default=utcnow)
+
+
 class FileFailure(Base):
     """Fehlversuche pro Datei, Grundlage für die Quarantäne."""
 

src/netmover/mover/engine.py

@@ -19,6 +19,7 @@ import posixpath
 import time
 from collections.abc import Callable
 from dataclasses import dataclass, field
+from datetime import datetime
 from pathlib import Path
 from typing import Any
 
@@ -72,6 +73,7 @@ class RuleSnapshot:
     set_mode: int | None = None
     quarantine_after: int = 3
     dst_root_squash: bool | None = None  # nur NFS-Ziele: Ergebnis des Schreibtests
+    archive_layout: str = "flat"  # "flat" | "date" (Jahr/Monat/…)
 
 
 @dataclass
@@ -159,6 +161,7 @@ class Engine:
         dry_run: bool = False,
         quarantine_rel: str | None = None,
         quarantine_root: Path | None = None,
+        archive: Destination | None = None,
     ) -> RunResult:
         """``source``: eine :class:`Source` oder (Kurzform) ein lokales Verzeichnis.
         Quarantäne relativ zur Wurzel der Anbindung (``quarantine_rel``) oder – bei
@@ -168,7 +171,7 @@ class Engine:
         if quarantine_root is not None and isinstance(src, LocalSource):
             quarantine_rel = str(quarantine_root.relative_to(src.root))
         try:
-            self._execute(rule, src, dest, token, result, dry_run, quarantine_rel)
+            self._execute(rule, src, dest, token, result, dry_run, quarantine_rel, archive)
         except Cancelled:
             result.status = RunStatus.ERROR
             result.message = "Lauf abgebrochen"
@@ -195,10 +198,13 @@ class Engine:
         result: RunResult,
         dry_run: bool,
         quarantine_rel: str | None,
+        archive: Destination | None = None,
     ) -> None:
         src.check_root()
         if not dry_run:
             dest.prepare()
+            if archive is not None:
+                archive.prepare()
         token.check()
 
         exclude = frozenset({quarantine_rel}) if quarantine_rel else frozenset()
@@ -212,8 +218,12 @@ class Engine:
         # Kopier-Buchführung (Fallstrick 11). Gilt auch für „verschieben“, falls
         # dort einmal die Quelle nicht gelöscht werden konnte.
         done = self.ledger.copied_keys(rule.id)
+        archived = self.ledger.archived_keys(rule.id) if archive is not None else set()
         if not dry_run:
-            self.ledger.prune_copied(rule.id, {c.relpath for c in candidates})
+            present = {c.relpath for c in candidates}
+            self.ledger.prune_copied(rule.id, present)
+            if archive is not None:
+                self.ledger.prune_archived(rule.id, present)
         fresh = [c for c in candidates if (c.relpath, c.state.size, c.state.mtime_ns) not in done]
         unchanged = len(candidates) - len(fresh)
 
@@ -271,7 +281,11 @@ class Engine:
         failures = self.ledger.failures(rule.id)
         for c in stable:
             token.check()
-            result.add(self._process(rule, c, src, dest, token, dry_run, quarantine_rel, failures))
+            result.add(
+                self._process(
+                    rule, c, src, dest, token, dry_run, quarantine_rel, failures, archive, archived
+                )
+            )
 
         if unchanged:
             result.message = f"{unchanged} bereits übertragen"
@@ -309,8 +323,14 @@ class Engine:
         dry_run: bool,
         quarantine_rel: str | None,
         failures: dict[str, FailureInfo],
+        archive: Destination | None = None,
+        archived: set[tuple[str, int, int]] | None = None,
     ) -> ItemResult:
         size = c.state.size
+        arch_rel = self._archive_rel(c.relpath, rule.archive_layout)
+        need_archive = archive is not None and (
+            (c.relpath, size, c.state.mtime_ns) not in (archived or set())
+        )
         failure = failures.get(c.relpath)
 
         # Quarantäne für wiederholt scheiternde Dateien
@@ -349,9 +369,14 @@ class Engine:
                 else None
             )
             verb = "kopieren" if rule.action is RuleAction.COPY else "verschieben"
-            return ItemResult(
+            msg = msg or f"würde {verb}"
-                c.relpath, ItemStatus.PLANNED, dest.describe(planned), size, msg or f"würde {verb}"
+            if archive is not None:
-            )
+                if need_archive:
+                    arch_planned = archive.plan(arch_rel, CollisionPolicy.RENAME)
+                    msg += f"; Archiv: {archive.describe(arch_planned or arch_rel)}"
+                else:
+                    msg += "; bereits archiviert"
+            return ItemResult(c.relpath, ItemStatus.PLANNED, dest.describe(planned), size, msg)
 
         try:
             free = dest.free_bytes()
@@ -367,10 +392,51 @@ class Engine:
                 f"benötigt {_mb(int(size * SPACE_RESERVE))})",
             )
 
+        if need_archive and archive is not None:
+            try:
+                afree = archive.free_bytes()
+            except OSError:
+                afree = None
+            if afree is not None and afree < size * SPACE_RESERVE:
+                return ItemResult(
+                    c.relpath,
+                    ItemStatus.FAILED,
+                    archive.describe(arch_rel),
+                    size,
+                    f"zu wenig Platz im Archiv (frei {_mb(afree)}, "
+                    f"benötigt {_mb(int(size * SPACE_RESERVE))})",
+                )
+
         def verify() -> None:
             if src.state(c.relpath) != c.state:
                 raise SourceChanged("Quelle hat sich während der Übertragung geändert")
 
+        # Archiv ZUERST: scheitert es, ist noch nichts ans Ziel (z. B. Paperless)
+        # geliefert. Scheitert danach das Ziel, verhindert die Buchführung eine
+        # zweite Archivkopie beim nächsten Versuch.
+        archive_note: str | None = None
+        if need_archive and archive is not None:
+            try:
+                meta, _ = self._meta(rule, src, c.relpath, dest)
+                ares = archive.commit(
+                    _SourceFile(src, c.relpath),
+                    size,
+                    arch_rel,
+                    CollisionPolicy.RENAME,
+                    MetaOptions(meta.atime_ns, meta.mtime_ns, meta.mode, None),
+                    verify,
+                    token,
+                )
+            except CollisionSkip:  # kann bei RENAME nicht auftreten
+                return ItemResult(c.relpath, ItemStatus.SKIPPED, None, size, "Archiv belegt")
+            except SourceChanged as exc:
+                return ItemResult(c.relpath, ItemStatus.SKIPPED, None, size, str(exc))
+            except (TransferError, OSError) as exc:
+                return self._fail(rule, c, exc, dry_run, prefix="Archiv: ")
+            archive_path = archive.describe(ares.final_rel)
+            self.ledger.record_archive(rule.id, c.relpath, size, c.state.mtime_ns, archive_path)
+            archive_note = f"archiviert: {archive_path}"
+
         try:
             meta, notes = self._meta(rule, src, c.relpath, dest)
             res = None
@@ -401,6 +467,8 @@ class Engine:
             return self._fail(rule, c, exc, dry_run)
 
         notes += res.notes
+        if archive_note:
+            notes.insert(0, archive_note)
         target = dest.describe(res.final_rel)
         if c.relpath in failures:
             self.ledger.clear_failure(rule.id, c.relpath)
@@ -425,10 +493,12 @@ class Engine:
                 )
         return ItemResult(c.relpath, ItemStatus.MOVED, target, size, "; ".join(notes) or None)
 
-    def _fail(self, rule: RuleSnapshot, c: Candidate, exc: Exception, dry_run: bool) -> ItemResult:
+    def _fail(
+        self, rule: RuleSnapshot, c: Candidate, exc: Exception, dry_run: bool, prefix: str = ""
+    ) -> ItemResult:
         if isinstance(exc, OSError) and exc.errno in MOUNT_ERRNOS:
-            raise RunAborted(f"Anbindung gestört: {_describe_os_error(exc)}") from exc
+            raise RunAborted(f"{prefix}Anbindung gestört: {_describe_os_error(exc)}") from exc
-        msg = _describe_os_error(exc) if isinstance(exc, OSError) else str(exc)
+        msg = prefix + (_describe_os_error(exc) if isinstance(exc, OSError) else str(exc))
         if not dry_run:
             attempts = self.ledger.record_failure(
                 rule.id, c.relpath, c.state.size, c.state.mtime_ns, msg
@@ -437,6 +507,12 @@ class Engine:
         event(log, "file.failed", level=30, rule=rule.name, file=c.relpath, error=msg)
         return ItemResult(c.relpath, ItemStatus.FAILED, None, c.state.size, msg)
 
+    def _archive_rel(self, relpath: str, layout: str) -> str:
+        if layout == "date":
+            t = datetime.fromtimestamp(self._clock())
+            return f"{t:%Y}/{t:%m}/{relpath}"
+        return relpath
+
     def _quarantine(
         self, rule: RuleSnapshot, c: Candidate, src: Source, q_rel: str, attempts: int
     ) -> ItemResult:

src/netmover/mover/ledger.py

@@ -14,7 +14,7 @@ from sqlalchemy import delete, select
 from sqlalchemy.orm import Session, sessionmaker
 
 from netmover.db import session_scope
-from netmover.models import CopiedFile, FileFailure, utcnow
+from netmover.models import ArchivedFile, CopiedFile, FileFailure, utcnow
 
 CopyKey = tuple[str, int, int]  # (relpath, size, mtime_ns)
 
@@ -72,6 +72,53 @@ class Ledger:
                 s.execute(delete(CopiedFile).where(CopiedFile.id.in_(stale)))
             return len(stale)
 
+    # -- Archiv ------------------------------------------------------------------------
+
+    def archived_keys(self, rule_id: int) -> set[CopyKey]:
+        with session_scope(self.session_factory) as s:
+            rows = s.execute(
+                select(ArchivedFile.src_relpath, ArchivedFile.size, ArchivedFile.mtime_ns).where(
+                    ArchivedFile.rule_id == rule_id
+                )
+            ).all()
+        return {(r[0], r[1], r[2]) for r in rows}
+
+    def record_archive(
+        self, rule_id: int, relpath: str, size: int, mtime_ns: int, archive_path: str
+    ) -> None:
+        with session_scope(self.session_factory) as s:
+            exists = s.scalar(
+                select(ArchivedFile.id).where(
+                    ArchivedFile.rule_id == rule_id,
+                    ArchivedFile.src_relpath == relpath,
+                    ArchivedFile.size == size,
+                    ArchivedFile.mtime_ns == mtime_ns,
+                )
+            )
+            if exists is None:
+                s.add(
+                    ArchivedFile(
+                        rule_id=rule_id,
+                        src_relpath=relpath,
+                        size=size,
+                        mtime_ns=mtime_ns,
+                        archive_path=archive_path,
+                    )
+                )
+
+    def prune_archived(self, rule_id: int, present: set[str]) -> int:
+        """Wie :meth:`prune_copied`: nur nach einem vollständigen Scan aufrufen."""
+        with session_scope(self.session_factory) as s:
+            rows = s.execute(
+                select(ArchivedFile.id, ArchivedFile.src_relpath).where(
+                    ArchivedFile.rule_id == rule_id
+                )
+            ).all()
+            stale = [r[0] for r in rows if r[1] not in present]
+            if stale:
+                s.execute(delete(ArchivedFile).where(ArchivedFile.id.in_(stale)))
+            return len(stale)
+
     def failures(self, rule_id: int) -> dict[str, FailureInfo]:
         with session_scope(self.session_factory) as s:
             rows = s.scalars(select(FileFailure).where(FileFailure.rule_id == rule_id)).all()

src/netmover/mover/runner.py

@@ -56,6 +56,11 @@ class RulePlan:
     quarantine_path: str | None
     op_timeout: int
     dry_run: bool
+    archive_kind: str = "none"
+    archive_mount_id: int | None = None
+    archive_type: MountType | None = None
+    archive_name: str | None = None
+    archive_path: str = ""
 
 
 def parse_owner(value: str | None) -> tuple[int, int] | None:
@@ -92,6 +97,7 @@ def load_plan(session: Session, rule_id: int) -> RulePlan:
         set_mode=parse_mode(r.set_mode),
         quarantine_after=r.quarantine_after,
         dst_root_squash=dst.root_squash if dst is not None else None,
+        archive_layout=r.archive_layout.value,
     )
     return RulePlan(
         snapshot=snap,
@@ -109,6 +115,11 @@ def load_plan(session: Session, rule_id: int) -> RulePlan:
         else None,
         op_timeout=r.op_timeout_seconds,
         dry_run=r.dry_run,
+        archive_kind=r.archive_kind.value,
+        archive_mount_id=r.archive_mount_id if r.archive_kind.value == "mount" else None,
+        archive_type=r.archive_mount.type if r.archive_mount is not None else None,
+        archive_name=r.archive_mount.name if r.archive_mount is not None else None,
+        archive_path=r.archive_path,
     )
 
 
@@ -176,24 +187,52 @@ class RuleRunner:
                 if plan.dst_mount_id is not None
                 else None
             )
+            arch_mp = (
+                self.mounts.check_alive(plan.archive_mount_id)
+                if plan.archive_mount_id is not None
+                else None
+            )
         except MountUnavailable as exc:
             return RunResult(RunStatus.ERROR, dry, message=str(exc), unavailable=True)
 
         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)
+            make_dest = self._target(
+                "dst_path",
+                plan.dst_mount_id,
+                plan.dst_type,
+                plan.dst_name,
+                plan.dst_path,
+                dst_mp,
+                src_root,
+            )
+            make_archive = (
+                self._target(
+                    "archive_path",
+                    plan.archive_mount_id,
+                    plan.archive_type,
+                    plan.archive_name,
+                    plan.archive_path,
+                    arch_mp,
+                    src_root,
+                )
+                if plan.archive_kind != "none"
+                else None
+            )
         except (ValidationError, TransferError, SftpError) as exc:
             return RunResult(RunStatus.ERROR, dry, message=str(exc))
 
         def work(token: CancelToken) -> RunResult:
+            opened: list[Source | Destination] = []
+            try:
                 source = make_source()
-            try:
+                opened.append(source)
                 dest = make_dest()
-            except BaseException:
+                opened.append(dest)
-                source.close()
+                archive = make_archive() if make_archive is not None else None
-                raise
+                if archive is not None:
-            try:
+                    opened.append(archive)
                 return self.engine.execute(
                     plan.snapshot,
                     source,
@@ -201,10 +240,11 @@ class RuleRunner:
                     token,
                     dry_run=dry,
                     quarantine_rel=plan.quarantine_path,
+                    archive=archive,
                 )
             finally:
-                dest.close()
+                for item in reversed(opened):
-                source.close()
+                    item.close()
 
         def finish() -> None:
             self._started.pop(rule_id, None)
@@ -251,41 +291,50 @@ class RuleRunner:
 
         return make_local
 
-    def _dest_factory(
+    def _target(
-        self, plan: RulePlan, dst_mp: Path | None, src_root: Path | None
+        self,
+        field: str,
+        mount_id: int | None,
+        mount_type: MountType | None,
+        mount_name: str | None,
+        path: str,
+        mp: Path | None,
+        src_root: Path | None,
     ) -> Callable[[], Destination]:
-        if plan.dst_mount_id is None:
+        """Ziel oder Archiv: lokaler Pfad (``mount_id`` None), SFTP oder Kernel-Mount."""
-            root = Path(safe_abspath("dst_path", plan.dst_path))
+        what = "Archiv" if field == "archive_path" else "Ziel"
+        if mount_id is None:
+            root = Path(safe_abspath(field, path))
             if self._under_mount_base(root):
                 base = self.mounts.mount_base
                 raise ValidationError(
-                    "dst_path",
+                    field,
-                    f"Lokales Ziel liegt unter {base} – bitte die Anbindung als Ziel wählen, "
+                    f"Lokales {what} liegt unter {base} – bitte die Anbindung als {what} wählen, "
                     "damit geprüft wird, ob sie eingehängt ist.",
                 )
             kind = "local"
             local_root = root
             create = False
-        elif plan.dst_type is MountType.SFTP:
+        elif mount_type is MountType.SFTP:
-            scfg = self.mounts.sftp_config_for(plan.dst_mount_id)
+            scfg = self.mounts.sftp_config_for(mount_id)
-            sub = safe_relpath("dst_path", plan.dst_path)
+            sub = safe_relpath(field, path)
             remote = posixpath.join(scfg.base_path, sub) if sub else scfg.base_path
-            label = plan.dst_name or "sftp"
+            label = mount_name or "sftp"
 
             def make_sftp() -> Destination:
                 return SftpDestination(connect(scfg), remote, label=label)
 
             return make_sftp
         else:
-            if dst_mp is None:
+            if mp is None:
-                raise TransferError("Ziel-Anbindung ist nicht eingehängt")
+                raise TransferError(f"{what}-Anbindung ist nicht eingehängt")
-            sub = safe_relpath("dst_path", plan.dst_path)
+            sub = safe_relpath(field, path)
-            local_root = dst_mp / sub if sub else dst_mp
+            local_root = mp / sub if sub else mp
-            kind = "cifs" if plan.dst_type is MountType.CIFS else "nfs"
+            kind = "cifs" if mount_type is MountType.CIFS else "nfs"
             create = True
 
         if src_root is not None and local_root.resolve() == src_root.resolve():
-            raise ValidationError("dst_path", "Quelle und Ziel sind derselbe Ordner")
+            raise ValidationError(field, f"Quelle und {what} sind derselbe Ordner")
 
         def make_local() -> Destination:
             return LocalDestination(local_root, kind, create_root=create)  # type: ignore[arg-type]

src/netmover/settings_store.py

@@ -46,9 +46,12 @@ class AppSettings:
     smtp_password_enc: str = ""
     smtp_from: str = ""
     smtp_to: str = ""
-    # Anmeldung: "always" | "trusted" (ohne Anmeldung aus trusted_networks) | "off"
+    # Anmeldung: Netze ohne Passwort (trusted_networks), Netze mit Passwort
+    # (allowed_networks, leer = alle übrigen); alle anderen Netze werden abgewiesen.
+    # auth_mode "off" = ohne Anmeldung von überall (aus 0.5.0, weiterhin gültig).
     auth_mode: str = "always"
     trusted_networks: str = ""
+    allowed_networks: str = ""
 
 
 _PREFIX = "app."

src/netmover/transfer.py

@@ -16,6 +16,8 @@ from sqlalchemy.orm import Session
 
 from netmover import __version__
 from netmover.models import (
+    ArchiveKind,
+    ArchiveLayout,
     CollisionPolicy,
     Mount,
     MountState,
@@ -72,6 +74,9 @@ RULE_FIELDS = (
     "dry_run",
     "quarantine_path",
     "quarantine_after",
+    "archive_kind",
+    "archive_path",
+    "archive_layout",
 )
 SETTING_FIELDS = (
     "retention_days",
@@ -99,6 +104,7 @@ def export_config(s: Session, settings: AppSettings) -> dict[str, Any]:
                 **{f: _plain(getattr(r, f)) for f in RULE_FIELDS},
                 "src_mount": names[r.src_mount_id],
                 "dst_mount": names.get(r.dst_mount_id) if r.dst_mount_id else None,
+                "archive_mount": names.get(r.archive_mount_id) if r.archive_mount_id else None,
             }
             for r in s.scalars(select(Rule).order_by(Rule.name))
         ],
@@ -165,13 +171,16 @@ def import_config(s: Session, mounts: MountManager, data: dict[str, Any]) -> Imp
         try:
             src = ids[raw["src_mount"]]
             dst = ids[raw["dst_mount"]] if raw.get("dst_mount") else None
-            r = Rule(src_mount_id=src, dst_mount_id=dst)
+            arch = ids[raw["archive_mount"]] if raw.get("archive_mount") else None
+            r = Rule(src_mount_id=src, dst_mount_id=dst, archive_mount_id=arch)
             for f in RULE_FIELDS:
                 if f in raw:
                     setattr(r, f, raw[f])
             r.pattern_type = PatternType(r.pattern_type)
             r.action = RuleAction(r.action)
             r.on_collision = CollisionPolicy(r.on_collision)
+            r.archive_kind = ArchiveKind(r.archive_kind or "none")
+            r.archive_layout = ArchiveLayout(r.archive_layout or "flat")
             r.src_subpath = safe_relpath("src_subpath", r.src_subpath or "")
             if int(r.interval_seconds) < 10:
                 raise ValueError("Intervall < 10 s")

src/netmover/web/app.py

@@ -38,6 +38,7 @@ from netmover.admin import (
     AdminError,
     MountInput,
     RuleInput,
+    copy_name,
     credential_for_test,
     delete_mount,
     mount_input_from_model,
@@ -269,7 +270,7 @@ async def guard(request: Request) -> None:
     if not auth.password_is_set(state.session_factory):
         raise auth.LoginRequired(setup=True)
     cfg = load_settings(state.session_factory)
-    bypass = auth.login_bypass(request, cfg.auth_mode, cfg.trusted_networks)
+    bypass = auth.login_bypass(request, cfg.auth_mode, cfg.trusted_networks, cfg.allowed_networks)
     if bypass:
         request.state.auth_bypass = bypass
     elif not auth.is_logged_in(request):
@@ -364,7 +365,7 @@ async def login_form(request: Request) -> Response:
     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):
+    if auth.login_bypass(request, cfg.auth_mode, cfg.trusted_networks, cfg.allowed_networks):
         return redirect("/")
     return render(request, "login.html")
 
@@ -422,13 +423,24 @@ class RuleView:
     target: str
 
 
+def _archive_target(r: Rule) -> str | None:
+    if r.archive_kind.value == "none":
+        return None
+    if r.archive_kind.value == "local" or r.archive_mount is None:
+        place = r.archive_path
+    else:
+        place = f"{r.archive_mount.name}:/{r.archive_path}"
+    return place + (" (nach Datum)" if r.archive_layout.value == "date" else "")
+
+
 def _rule_target(r: Rule) -> str:
     src = f"{r.src_mount.name}:/{r.src_subpath}" if r.src_subpath else f"{r.src_mount.name}:/"
     if r.dst_mount is None:
         dst = r.dst_path
     else:
         dst = f"{r.dst_mount.name}:/{r.dst_path}" if r.dst_path else f"{r.dst_mount.name}:/"
-    return f"{src} → {dst}"
+    archive = _archive_target(r)
+    return f"{src} → {dst}" + (f" + Archiv {archive}" if archive else "")
 
 
 async def _status_context(request: Request) -> dict[str, Any]:
@@ -580,10 +592,12 @@ def _mount_form(
     errors: dict[str, str],
     mount: Mount | None = None,
     status_code: int = 200,
+    copy_from: Mount | None = None,
 ) -> Response:
+    owner = mount or copy_from
     pub = (
-        mount.credential.public_key
+        owner.credential.public_key
-        if mount is not None and mount.credential and not mount.credential.managed
+        if owner is not None and owner.credential and not owner.credential.managed
         else None
     )
     return render(
@@ -596,8 +610,9 @@ def _mount_form(
         mount=mount,
         public_key=pub,
         has_secret=bool(
-            mount is not None and mount.credential is not None and not mount.credential.managed
+            owner is not None and owner.credential is not None and not owner.credential.managed
         ),
+        copy_from=copy_from,
         key_options=_key_options(request),
         password_options=_password_options(request),
     )
@@ -617,16 +632,33 @@ async def mount_edit(request: Request, mount_id: int) -> Response:
         return _mount_form(request, mount_input_from_model(m), {}, m)
 
 
+@protected.get("/mounts/{source_id}/copy", response_class=HTMLResponse)
+async def mount_copy(request: Request, source_id: int) -> Response:
+    """Formular für eine neue Anbindung, vorausgefüllt aus einer bestehenden."""
+    with nm(request).session_factory() as s:
+        src = s.get(Mount, source_id)
+        if src is None:
+            return redirect("/mounts")
+        data = mount_input_from_model(src)
+        data.name = copy_name(s, src.name)
+        return _mount_form(request, data, {}, copy_from=src)
+
+
 @protected.post("/mounts/new", response_class=HTMLResponse)
 @protected.post("/mounts/{mount_id}/edit", response_class=HTMLResponse)
-async def mount_save(request: Request, mount_id: int | None = None) -> Response:
+@protected.post("/mounts/{source_id}/copy", response_class=HTMLResponse)
+async def mount_save(
+    request: Request, mount_id: int | None = None, source_id: int | None = None
+) -> Response:
     state = nm(request)
     data, errors = parse_mount_form(_form_dict(await request.form()))
     with session_scope(state.session_factory) as s:
-        existing = s.get(Mount, mount_id) if mount_id is not None else None
+        copy_from = s.get(Mount, source_id) if source_id is not None else None
         if not errors:
             try:
-                m, errors = save_mount(s, state.secrets, state.mounts, data, mount_id)
+                m, errors = save_mount(
+                    s, state.secrets, state.mounts, data, mount_id, copy_from=copy_from
+                )
             except AdminError as exc:
                 errors = {"_": str(exc)}
                 m = None
@@ -635,7 +667,10 @@ async def mount_save(request: Request, mount_id: int | None = None) -> Response:
         if errors or m is None:
             s.rollback()
             existing = s.get(Mount, mount_id) if mount_id is not None else None
-            return _mount_form(request, data, errors, existing, status_code=400)
+            copy_from = s.get(Mount, source_id) if source_id is not None else None
+            return _mount_form(
+                request, data, errors, existing, status_code=400, copy_from=copy_from
+            )
         saved_id, name = m.id, m.name
         is_sftp_key = m.type is MountType.SFTP and data.auth is CredentialKind.SSH_KEY
     flash(request, f"Anbindung „{name}“ gespeichert.")
@@ -657,12 +692,17 @@ async def mount_test_form(request: Request) -> Response:
     data, errors = parse_mount_form(form)
     if errors:
         return render(request, "partials/check_result.html", errors=errors, report=None)
-    existing_id = int(form["mount_id"]) if form.get("mount_id", "").isdigit() else None
+    ref = form.get("mount_id") or form.get("copy_from") or ""
+    existing_id = int(ref) if ref.isdigit() else None
     with state.session_factory() as s:
         existing = s.get(Mount, existing_id) if existing_id else None
         cred = credential_for_test(data, existing, state.secrets, s)
     cfg = data.config()
-    if data.type is MountType.SFTP and existing is not None:
+    if (
+        data.type is MountType.SFTP
+        and existing is not None
+        and (existing.server, existing.port) == (data.server, data.port)
+    ):
         cfg = dataclasses.replace(cfg, host_key=existing.host_key)
     report = await run_in_threadpool(state.mounts.check, cfg, cred, None)
     return render(request, "partials/check_result.html", report=report, errors={})
@@ -1235,39 +1275,54 @@ async def settings_password(request: Request) -> Response:
 
 @protected.post("/settings/auth", response_class=HTMLResponse)
 async def settings_auth(request: Request) -> Response:
-    """Anmeldung abschalten bzw. auf vertrauenswürdige Netze beschränken.
+    """Netze ohne / mit Passwort festlegen. Nur mit dem aktuellen Passwort und nie so,
-    Nur mit dem aktuellen Passwort."""
+    dass man sich mit der eigenen Adresse aussperrt."""
     state = nm(request)
     form = _form_dict(await request.form())
+    back = redirect("/settings#anmeldung")
     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")
+        return back
-    mode = form.get("auth_mode", auth.AUTH_ALWAYS)
+    trusted = (form.get("trusted_networks") or "").strip()
-    networks = (form.get("trusted_networks") or "").strip()
+    allowed = (form.get("allowed_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)
+        auth.parse_networks(trusted)
+        auth.parse_networks(allowed)
     except ValueError as exc:
         flash(request, str(exc), "error")
-        return redirect("/settings#anmeldung")
+        return back
-    if mode == auth.AUTH_TRUSTED and not nets:
+    me = auth.client_host(request)
-        flash(request, "Bitte mindestens ein vertrauenswürdiges Netz angeben.", "error")
+    if auth.access_decision(me, auth.AUTH_ALWAYS, trusted, allowed) == auth.ACCESS_DENY:
-        return redirect("/settings#anmeldung")
+        flash(
+            request,
+            f"Nicht gespeichert: Damit wärst du selbst ausgesperrt (deine Adresse {me}). "
+            "Trage dein Netz bei „mit“ oder „ohne Passwort“ ein.",
+            "error",
+        )
+        return back
     cfg = load_settings(state.session_factory)
-    cfg.auth_mode, cfg.trusted_networks = mode, networks
+    cfg.trusted_networks, cfg.allowed_networks = trusted, allowed
+    cfg.auth_mode = auth.AUTH_TRUSTED if trusted else auth.AUTH_ALWAYS
     save_settings(state.session_factory, cfg)
-    event(log, "auth.mode_changed", level=30, mode=mode, networks=networks)
+    event(
-    msgs = {
+        log,
-        auth.AUTH_ALWAYS: "Anmeldung ist wieder immer erforderlich.",
+        "auth.access_changed",
-        auth.AUTH_TRUSTED: f"Ohne Anmeldung aus: {networks}. Von überall sonst mit Passwort.",
+        level=30,
-        auth.AUTH_OFF: "Anmeldung abgeschaltet – NetMover ist ohne Passwort bedienbar.",
+        without_password=trusted,
-    }
+        with_password=allowed or "alle übrigen",
-    flash(request, msgs[mode], "info" if mode == auth.AUTH_ALWAYS else "warn")
+    )
-    return redirect("/settings#anmeldung")
+    parts = [f"ohne Passwort aus: {trusted}" if trusted else "überall mit Passwort"]
+    parts.append(
+        f"mit Passwort nur aus: {allowed} – alle anderen Netze abgewiesen"
+        if allowed
+        else "alle übrigen Netze mit Passwort"
+    )
+    flash(
+        request, "Anmeldung gespeichert: " + "; ".join(parts) + ".", "warn" if trusted else "info"
+    )
+    return back
 
 
 @protected.get("/settings/export")
@@ -1371,6 +1426,35 @@ def create_app(settings: Settings | None = None, state: NetMover | None = None)
         openapi_url=None,
     )
 
+    @app.middleware("http")
+    async def network_access(
+        request: Request, call_next: Callable[[Request], Awaitable[Response]]
+    ) -> Response:
+        path = request.url.path
+        if not (path.startswith("/healthz") or path.startswith("/static/")):
+            app_state = getattr(request.app.state, "nm", None)
+            if app_state is not None:
+                cfg = load_settings(app_state.session_factory)
+                decision = auth.access_decision(
+                    auth.client_host(request),
+                    cfg.auth_mode,
+                    cfg.trusted_networks,
+                    cfg.allowed_networks,
+                )
+                if decision == auth.ACCESS_DENY:
+                    event(
+                        log,
+                        "auth.denied_network",
+                        level=30,
+                        client=auth.client_host(request) or "?",
+                    )
+                    return HTMLResponse(
+                        "<!doctype html><meta charset=utf-8><title>NetMover</title>"
+                        "<p>Zugriff aus diesem Netz ist nicht erlaubt.</p>",
+                        status_code=403,
+                    )
+        return await call_next(request)
+
     @app.middleware("http")
     async def security_headers(
         request: Request, call_next: Callable[[Request], Awaitable[Response]]

src/netmover/web/auth.py

@@ -51,25 +51,59 @@ def parse_networks(raw: str) -> list[ipaddress.IPv4Network | ipaddress.IPv6Netwo
     return nets
 
 
-def login_bypass(request: Request, mode: str, trusted: str) -> str | None:
+ACCESS_NOPW, ACCESS_PW, ACCESS_DENY = "nopw", "pw", "deny"
-    """Grund, warum keine Anmeldung nötig ist (``"off"``/``"trusted"``), sonst ``None``.
+
+
+def _address(host: str | None) -> ipaddress.IPv4Address | ipaddress.IPv6Address | None:
+    if not host:
+        return None
+    try:
+        return ipaddress.ip_address(host)
+    except ValueError:
+        return None
+
+
+def _networks_or_empty(raw: str) -> list[ipaddress.IPv4Network | ipaddress.IPv6Network]:
+    try:
+        return parse_networks(raw)
+    except ValueError:
+        return []  # gespeicherte Werte sind beim Speichern geprüft
+
+
+def access_decision(host: str | None, mode: str, trusted: str, allowed: str) -> str:
+    """Wie darf diese Adresse auf die Oberfläche zugreifen?
+
+    1. aus ``trusted`` (oder Modus „off“): **ohne Passwort**,
+    2. ``allowed`` leer oder Adresse darin: **mit Passwort**,
+    3. sonst: **abgewiesen**.
 
     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
+        return ACCESS_NOPW
-    if mode != AUTH_TRUSTED or request.client is None:
+    addr = _address(host)
+    if addr is not None and any(addr in n for n in _networks_or_empty(trusted)):
+        return ACCESS_NOPW
+    allowed_nets = _networks_or_empty(allowed)
+    if allowed_nets and (addr is None or not any(addr in n for n in allowed_nets)):
+        return ACCESS_DENY
+    return ACCESS_PW
+
+
+def client_host(request: Request) -> str | None:
+    return request.client.host if request.client else None
+
+
+def login_bypass(request: Request, mode: str, trusted: str, allowed: str = "") -> str | None:
+    """Grund, warum keine Anmeldung nötig ist (``"off"``/``"trusted"``), sonst ``None``."""
+    if access_decision(client_host(request), mode, trusted, allowed) != ACCESS_NOPW:
         return None
-    try:
+    return AUTH_OFF if mode == AUTH_OFF or _covers_everything(trusted) else AUTH_TRUSTED
-        addr = ipaddress.ip_address(request.client.host)
+
-    except ValueError:
+
-        return None
+def _covers_everything(raw: str) -> bool:
-    try:
+    return any(n.prefixlen == 0 for n in _networks_or_empty(raw))
-        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:

src/netmover/web/templates/mount_form.html

@@ -1,12 +1,14 @@
 {% extends "base.html" %}
 {% from "macros.html" import field, select, check %}
-{% block title %}{{ 'Anbindung bearbeiten' if mount else 'Neue Anbindung' }}{% endblock %}
+{% block title %}{{ 'Anbindung bearbeiten' if mount else ('Anbindung kopieren' if copy_from else 'Neue Anbindung') }}{% endblock %}
 {% block content %}
-<h1>{{ 'Anbindung „' ~ mount.name ~ '“ bearbeiten' if mount else 'Neue Anbindung' }}</h1>
+<h1>{% if mount %}Anbindung „{{ mount.name }}“ bearbeiten{% elif copy_from %}Kopie von „{{ copy_from.name }}“{% else %}Neue Anbindung{% endif %}</h1>
+{% if copy_from %}<p class="msg info small">Alle Werte stammen aus „{{ copy_from.name }}“. Zugangsdaten werden übernommen, solange du das Passwort leer lässt – die Kopie bekommt ihr eigenes Exemplar.{% if copy_from.credential and copy_from.credential.public_key and not copy_from.credential.managed %} Der SSH-Schlüssel bleibt derselbe, ein neuer Eintrag in <code>authorized_keys</code> ist nicht nötig.{% endif %}</p>{% endif %}
 {% if errors.get('_') %}<p class="msg error">{{ errors['_'] }}</p>{% endif %}
 <form class="form" method="post" data-dynamic>
   <input type="hidden" name="csrf_token" value="{{ csrf }}">
   {% if mount %}<input type="hidden" name="mount_id" value="{{ mount.id }}">{% endif %}
+  {% if copy_from %}<input type="hidden" name="copy_from" value="{{ copy_from.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')], d.type.value, errors=errors) }}

src/netmover/web/templates/mounts.html

@@ -22,6 +22,7 @@
         {% endif %}
       {% endif %}
       <a class="btn small" href="/mounts/{{ m.id }}/edit">Bearbeiten</a>
+      <a class="btn small" href="/mounts/{{ m.id }}/copy" title="Neue Anbindung mit denselben Werten und Zugangsdaten">Kopieren</a>
       <form class="inline" method="post" action="/mounts/{{ m.id }}/delete" data-confirm="„{{ m.name }}“ wirklich löschen?">{{ csrf_field(csrf) }}<button class="small danger">Löschen</button></form>
     </div>
   </div>

src/netmover/web/templates/rule_form.html

@@ -30,6 +30,13 @@
     {{ select('on_collision', 'Wenn Datei existiert', [('rename', 'umbenennen (scan-1.pdf)'), ('skip', 'überspringen'), ('overwrite', 'überschreiben')], d.on_collision.value, errors=errors) }}
   </div></fieldset>
 
+  <fieldset><legend>Archiv <a class="legend-help" href="/help#archiv" title="Hilfe">?</a></legend><div class="fields">
+    {{ select('archive_kind', 'Archiv', [('none', 'kein Archiv'), ('mount', 'Anbindung'), ('local', 'Lokaler Pfad')], d.archive_kind.value, errors=errors, hint='Zusätzliche Kopie. Die Quelle wird erst gelöscht, wenn Ziel UND Archiv sie haben.') }}
+    <div data-show-for="archive_kind:mount">{{ select('archive_mount_id', 'Archiv-Anbindung', dst_options, d.archive_mount_id or '', errors=errors) }}</div>
+    <div data-show-for="archive_kind:mount local">{{ field('archive_path', 'Archivordner', d.archive_path, errors=errors, hint='Anbindung: relativer Unterordner (wird angelegt). Lokal: absoluter Pfad, muss existieren.') }}</div>
+    <div data-show-for="archive_kind:mount local">{{ select('archive_layout', 'Ablage', [('flat', 'wie in der Quelle'), ('date', 'nach Datum (Jahr/Monat)')], d.archive_layout.value, errors=errors, hint='Im Archiv wird nie überschrieben – gleiche Namen werden umbenannt.') }}</div>
+  </div></fieldset>
+
   <fieldset><legend>Sicherheit und Grenzen <a class="legend-help" href="/help#regel-sicherheit" title="Hilfe">?</a></legend><div class="fields">
     {{ field('stability_seconds', 'Stabilitätsfenster (s)', d.stability_seconds, type='number', errors=errors, attrs='min="0"', hint='Größe und Zeit müssen so lange gleich bleiben') }}
     {{ field('min_age_seconds', 'Mindestalter (s)', d.min_age_seconds, type='number', errors=errors, attrs='min="0"') }}

src/netmover/web/templates/settings.html

@@ -66,20 +66,23 @@
 </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>
+<form class="form card stack" method="post" action="/settings/auth">
   {{ csrf_field(csrf) }}
+  <p class="small muted">Deine Adresse gerade: <code>{{ client_ip }}</code></p>
   <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"><label for="f-nets">Ohne Passwort aus diesen Netzen</label>
-    <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="z. B. 10.0.10.0/24, 10.1.5.0/24">
-      <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; einzelne Adressen gehen auch. Leer = niemand ohne Passwort. <code>0.0.0.0/0, ::/0</code> = Anmeldung ganz aus.</span></div>
-      <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 wide"><label for="f-allowed">Mit Passwort aus diesen Netzen</label>
+      <input id="f-allowed" type="text" name="allowed_networks" value="{{ cfg.allowed_networks }}" placeholder="leer = alle übrigen Netze">
+      <span class="hint">Leer = von überall sonst mit Passwort. Ausgefüllt = <strong>alle anderen Netze werden abgewiesen</strong> (auch die Anmeldeseite). <code>/healthz</code> bleibt für die Überwachung erreichbar.</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.
+  <p class="msg warn small">NetMover läuft als root und kennt die Zugangsdaten deiner Freigaben. Wer die Oberfläche ohne Passwort erreicht,
-    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.
+    kann alles ändern – Netze „ohne Passwort“ daher so eng wie möglich wählen. Hinter einem Reverse Proxy <code>NETMOVER_FORWARDED_ALLOW_IPS</code> setzen,
-    Hinter einem Proxy außerdem <code>NETMOVER_FORWARDED_ALLOW_IPS</code> setzen, sonst sieht jede Anfrage aus, als käme sie vom Proxy.</p>
+    sonst sieht jede Anfrage aus, als käme sie vom Proxy. Eine Einstellung, mit der du dich selbst aussperren würdest, wird nicht gespeichert.</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 %}
+  {% if loopback_trusted %}<p class="msg warn small">„Ohne Passwort“ 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>
 

tests/test_archive.py

@@ -0,0 +1,351 @@
+"""0.7.0: Archiv – Kopie ans Ziel UND ins Archiv, Quelle erst danach löschen."""
+
+from __future__ import annotations
+
+import os
+from collections.abc import Iterator
+from datetime import datetime
+from pathlib import Path
+
+import pytest
+from sqlalchemy import select
+from sqlalchemy.orm import Session, sessionmaker
+
+from netmover.db import session_scope
+from netmover.models import (
+    ArchivedFile,
+    ArchiveKind,
+    CollisionPolicy,
+    ItemStatus,
+    Rule,
+    RuleAction,
+    RunStatus,
+)
+from netmover.mounting.fake import FakeMountBackend
+from netmover.mounting.manager import MountManager
+from netmover.mover.destinations import LocalDestination, TransferError
+from netmover.mover.engine import Engine, RuleSnapshot, RunResult
+from netmover.mover.ledger import Ledger
+from netmover.mover.runner import RuleRunner
+from netmover.timeouts import CancelToken
+from tests.conftest import MountFactory
+
+
+@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
+
+
+FIXED = datetime(2026, 10, 12, 9, 30).timestamp()
+
+
+@pytest.fixture
+def engine(session_factory: sessionmaker[Session]) -> Engine:
+    return Engine(Ledger(session_factory), sleep=lambda _: None, clock=lambda: FIXED)
+
+
+@pytest.fixture
+def dirs(tmp_path: Path) -> tuple[Path, Path, Path]:
+    src, dst, arch = tmp_path / "src", tmp_path / "dst", tmp_path / "archiv"
+    for d in (src, dst, arch):
+        d.mkdir()
+    return src, dst, arch
+
+
+@pytest.fixture(autouse=True)
+def cross_device(monkeypatch: pytest.MonkeyPatch) -> None:
+    """tmp_path liegt auf einem Gerät – realistischer Weg über Kopie erzwingen."""
+    monkeypatch.setattr(LocalDestination, "move_same_device", lambda *a, **k: None)
+
+
+def snap(rule_id: int, **kw: object) -> RuleSnapshot:
+    return RuleSnapshot(id=rule_id, name="r", stability_seconds=0, **kw)  # type: ignore[arg-type]
+
+
+def run(
+    engine: Engine, rule: RuleSnapshot, src: Path, dst: Path, arch: Path | None, **kw: object
+) -> RunResult:
+    archive = LocalDestination(arch) if arch is not None else None
+    return engine.execute(rule, src, LocalDestination(dst), CancelToken(), archive=archive, **kw)  # type: ignore[arg-type]
+
+
+def write(p: Path, data: bytes = b"inhalt") -> Path:
+    p.parent.mkdir(parents=True, exist_ok=True)
+    p.write_bytes(data)
+    return p
+
+
+def names(d: Path) -> list[str]:
+    return sorted(str(p.relative_to(d)) for p in d.rglob("*") if p.is_file())
+
+
+def test_move_delivers_to_target_and_archive_then_deletes_source(
+    engine: Engine, rule_id: int, dirs: tuple[Path, Path, Path]
+) -> None:
+    src, dst, arch = dirs
+    data = os.urandom(200_000)
+    write(src / "scan001.pdf", data)
+    result = run(engine, snap(rule_id), src, dst, arch)
+    assert result.status is RunStatus.OK
+    item = result.items[0]
+    assert item.status is ItemStatus.MOVED and item.message and "archiviert:" in item.message
+    assert (dst / "scan001.pdf").read_bytes() == data == (arch / "scan001.pdf").read_bytes()
+    assert not (src / "scan001.pdf").exists()
+    assert names(dst) == ["scan001.pdf"] and names(arch) == ["scan001.pdf"]  # keine .part-Reste
+
+
+def test_date_layout(engine: Engine, rule_id: int, dirs: tuple[Path, Path, Path]) -> None:
+    src, dst, arch = dirs
+    write(src / "max" / "a.pdf")
+    run(engine, snap(rule_id, recursive=True, archive_layout="date"), src, dst, arch)
+    assert names(arch) == ["2026/10/max/a.pdf"] and names(dst) == ["max/a.pdf"]
+
+
+def test_archive_failure_delivers_nothing_and_keeps_source(
+    engine: Engine, rule_id: int, dirs: tuple[Path, Path, Path], monkeypatch: pytest.MonkeyPatch
+) -> None:
+    src, dst, arch = dirs
+    write(src / "a.pdf")
+    real = LocalDestination._link_noreplace
+
+    def fail_in_archive(self: LocalDestination, tmp: str, final: str) -> None:
+        if self.root == arch:
+            raise TransferError("Archiv voll")
+        real(self, tmp, final)
+
+    monkeypatch.setattr(LocalDestination, "_link_noreplace", fail_in_archive)
+    result = run(engine, snap(rule_id), src, dst, arch)
+    item = result.items[0]
+    assert item.status is ItemStatus.FAILED and item.message == "Archiv: Archiv voll (Versuch 1)"
+    assert (src / "a.pdf").exists() and names(dst) == [] and names(arch) == []
+
+
+def test_target_failure_after_archive_never_archives_twice(
+    engine: Engine,
+    rule_id: int,
+    dirs: tuple[Path, Path, Path],
+    monkeypatch: pytest.MonkeyPatch,
+    session_factory: sessionmaker[Session],
+) -> None:
+    src, dst, arch = dirs
+    write(src / "a.pdf")
+    real = LocalDestination._link_noreplace
+
+    def fail_in_target(self: LocalDestination, tmp: str, final: str) -> None:
+        if self.root == dst:
+            raise TransferError("Ziel gesperrt")
+        real(self, tmp, final)
+
+    monkeypatch.setattr(LocalDestination, "_link_noreplace", fail_in_target)
+    first = run(engine, snap(rule_id), src, dst, arch)
+    assert first.items[0].status is ItemStatus.FAILED
+    assert (src / "a.pdf").exists() and names(arch) == ["a.pdf"] and names(dst) == []
+    with session_factory() as s:
+        assert len(s.scalars(select(ArchivedFile)).all()) == 1
+
+    monkeypatch.setattr(LocalDestination, "_link_noreplace", real)  # Ziel wieder frei
+    second = run(engine, snap(rule_id), src, dst, arch)
+    assert second.items[0].status is ItemStatus.MOVED
+    assert names(arch) == ["a.pdf"]  # keine zweite Archivkopie (a-1.pdf)
+    assert names(dst) == ["a.pdf"] and not (src / "a.pdf").exists()
+
+
+def test_archive_never_overwrites(
+    engine: Engine, rule_id: int, dirs: tuple[Path, Path, Path]
+) -> None:
+    src, dst, arch = dirs
+    write(arch / "scan.pdf", b"alt im Archiv")
+    write(src / "scan.pdf", b"neu")
+    run(engine, snap(rule_id, on_collision=CollisionPolicy.OVERWRITE), src, dst, arch)
+    assert (arch / "scan.pdf").read_bytes() == b"alt im Archiv"
+    assert (arch / "scan-1.pdf").read_bytes() == b"neu"
+
+
+def test_no_space_in_archive_writes_nothing(
+    engine: Engine, rule_id: int, dirs: tuple[Path, Path, Path], monkeypatch: pytest.MonkeyPatch
+) -> None:
+    src, dst, arch = dirs
+    write(src / "a.pdf", b"x" * 1000)
+    real = LocalDestination.free_bytes
+    monkeypatch.setattr(
+        LocalDestination, "free_bytes", lambda self: 10 if self.root == arch else real(self)
+    )
+    result = run(engine, snap(rule_id), src, dst, arch)
+    assert "zu wenig Platz im Archiv" in (result.items[0].message or "")
+    assert names(dst) == [] and names(arch) == [] and (src / "a.pdf").exists()
+
+
+def test_source_changed_during_archive_copy(
+    engine: Engine, rule_id: int, dirs: tuple[Path, Path, Path], monkeypatch: pytest.MonkeyPatch
+) -> None:
+    src, dst, arch = dirs
+    f = write(src / "a.pdf", b"original")
+    real = LocalDestination._apply_meta
+
+    def change(self: LocalDestination, rel: str, meta: object) -> list[str]:
+        if self.root == arch:
+            f.write_bytes(b"geaendert!!")
+        return real(self, rel, meta)  # type: ignore[arg-type]
+
+    monkeypatch.setattr(LocalDestination, "_apply_meta", change)
+    result = run(engine, snap(rule_id), src, dst, arch)
+    assert result.items[0].status is ItemStatus.SKIPPED
+    assert names(dst) == [] and names(arch) == [] and f.exists()
+
+
+def test_copy_mode_with_archive(
+    engine: Engine, rule_id: int, dirs: tuple[Path, Path, Path]
+) -> None:
+    src, dst, arch = dirs
+    f = write(src / "a.pdf", b"v1")
+    rule = snap(rule_id, action=RuleAction.COPY)
+    for _ in range(3):
+        run(engine, rule, src, dst, arch)
+    assert f.exists() and names(dst) == ["a.pdf"] and names(arch) == ["a.pdf"]
+    f.write_bytes(b"version 2")
+    run(engine, rule, src, dst, arch)
+    assert names(dst) == ["a-1.pdf", "a.pdf"] and names(arch) == ["a-1.pdf", "a.pdf"]
+
+
+def test_dry_run_shows_archive_and_writes_nothing(
+    engine: Engine, rule_id: int, dirs: tuple[Path, Path, Path]
+) -> None:
+    src, dst, arch = dirs
+    write(src / "a.pdf")
+    result = run(engine, snap(rule_id, archive_layout="date"), src, dst, arch, dry_run=True)
+    assert result.items[0].status is ItemStatus.PLANNED
+    assert f"Archiv: {arch / '2026/10/a.pdf'}" in (result.items[0].message or "")
+    assert names(dst) == [] and names(arch) == [] and (src / "a.pdf").exists()
+
+
+def test_without_archive_unchanged(
+    engine: Engine, rule_id: int, dirs: tuple[Path, Path, Path]
+) -> None:
+    src, dst, arch = dirs
+    write(src / "a.pdf")
+    result = run(engine, snap(rule_id), src, dst, None)
+    assert result.items[0].message is None and names(arch) == []
+
+
+# -- Regellauf ----------------------------------------------------------------------------------
+
+
+@pytest.fixture
+def setup(
+    manager: MountManager,
+    fake_backend: FakeMountBackend,
+    make_mount: MountFactory,
+    session_factory: sessionmaker[Session],
+    tmp_path: Path,
+) -> Iterator[tuple[int, Path, Path, Path]]:
+    scans = fake_backend.export("nas", "scans")
+    arch_remote = fake_backend.export("nas", "archiv")
+    src_id = make_mount("scans", share="scans")
+    arch_id = make_mount("archiv", share="archiv")
+    manager.mount(src_id)
+    manager.mount(arch_id)
+    consume = tmp_path / "consume"
+    consume.mkdir()
+    with session_scope(session_factory) as s:
+        r = Rule(
+            name="Scans → Paperless + Archiv",
+            src_mount_id=src_id,
+            dst_path=str(consume),
+            stability_seconds=0,
+            archive_kind=ArchiveKind.MOUNT,
+            archive_mount_id=arch_id,
+            archive_path="scans",
+        )
+        s.add(r)
+        s.flush()
+        rid = r.id
+    yield rid, scans, consume, arch_remote
+
+
+def test_runner_with_archive_mount(
+    setup: tuple[int, Path, Path, Path],
+    manager: MountManager,
+    session_factory: sessionmaker[Session],
+) -> None:
+    rid, scans, consume, arch = setup
+    write(scans / "doc.pdf", b"%PDF")
+    result = RuleRunner(session_factory, manager, sleep=lambda _: None).run(rid)
+    assert result.status is RunStatus.OK, result.message
+    assert (
+        (consume / "doc.pdf").read_bytes() == b"%PDF" == (arch / "scans" / "doc.pdf").read_bytes()
+    )
+    assert not (scans / "doc.pdf").exists()
+
+
+def test_unavailable_archive_makes_rule_wait(
+    setup: tuple[int, Path, Path, Path],
+    manager: MountManager,
+    fake_backend: FakeMountBackend,
+    session_factory: sessionmaker[Session],
+) -> None:
+    rid, scans, consume, _arch = setup
+    write(scans / "doc.pdf")
+    fake_backend.vanish(manager.mountpoint_for("archiv"))
+    result = RuleRunner(session_factory, manager, sleep=lambda _: None).run(rid)
+    assert result.unavailable and result.status is RunStatus.ERROR
+    assert (scans / "doc.pdf").exists() and list(consume.iterdir()) == []
+
+
+def test_mount_used_as_archive_cannot_be_deleted(
+    setup: tuple[int, Path, Path, Path],
+    session_factory: sessionmaker[Session],
+    manager: MountManager,
+) -> None:
+    from netmover.admin import AdminError, delete_mount
+    from netmover.models import Mount
+
+    with session_scope(session_factory) as s:
+        arch_id = s.scalar(select(Mount.id).where(Mount.name == "archiv"))
+        assert arch_id is not None
+        with pytest.raises(AdminError, match="Wird noch von Regeln benutzt"):
+            delete_mount(s, manager, arch_id)
+
+
+def test_rule_form_archive_validation(
+    session_factory: sessionmaker[Session], make_mount: MountFactory
+) -> None:
+    from netmover.admin import parse_rule_form, save_rule
+
+    mid = make_mount()
+    base = {
+        "name": "x",
+        "src_mount_id": str(mid),
+        "dst_kind": "local",
+        "dst_path": "/srv/consume",
+        "interval_seconds": "30",
+        "stability_seconds": "5",
+        "op_timeout_seconds": "120",
+    }
+    with session_scope(session_factory) as s:
+        _, err = parse_rule_form(
+            s, {**base, "archive_kind": "local", "archive_path": "/srv/consume"}
+        )
+        assert "derselbe Ordner" in err["archive_path"]
+        _, err = parse_rule_form(s, {**base, "archive_kind": "mount", "archive_path": "a"})
+        assert "archive_mount_id" in err
+        _, err = parse_rule_form(s, {**base, "archive_kind": "local", "archive_path": "relativ"})
+        assert "absoluten Pfad" in err["archive_path"]
+        d, err = parse_rule_form(
+            s,
+            {
+                **base,
+                "archive_kind": "mount",
+                "archive_mount_id": str(mid),
+                "archive_path": "archiv/scans",
+                "archive_layout": "date",
+            },
+        )
+        assert err == {}
+        r, _ = save_rule(s, d)
+        assert r is not None and r.archive_kind is ArchiveKind.MOUNT
+        assert r.archive_path == "archiv/scans" and r.archive_layout.value == "date"

tests/test_auth_modes.py

@@ -1,4 +1,4 @@
-"""0.5.0: Anmeldung abschaltbar – immer / nicht aus vertrauenswürdigen Netzen / nie."""
+"""Anmeldung je Netz: ohne Passwort / mit Passwort / abgewiesen."""
 
 from __future__ import annotations
 
@@ -7,15 +7,16 @@ from typing import Any
 import pytest
 from fastapi.testclient import TestClient
 
-from netmover.settings_store import load_settings
+from netmover.settings_store import load_settings, save_settings
-from netmover.web.auth import includes_loopback, parse_networks
+from netmover.web.auth import (
-from tests.webutil import PASSWORD, csrf_of, post
+    ACCESS_DENY,
-
+    ACCESS_NOPW,
-
+    ACCESS_PW,
-def _set_mode(client: TestClient, mode: str, nets: str = "", current: str = PASSWORD) -> Any:
+    access_decision,
-    return post(
+    includes_loopback,
-        client, "/settings/auth", {"auth_mode": mode, "trusted_networks": nets, "current": current}
+    parse_networks,
 )
+from tests.webutil import PASSWORD, csrf_of
 
 
 def _other(client: TestClient, host: str) -> TestClient:
@@ -23,70 +24,124 @@ def _other(client: TestClient, host: str) -> TestClient:
     return TestClient(client.app, follow_redirects=False, client=(host, 51000))
 
 
-@pytest.fixture
+def _save(c: TestClient, trusted: str = "", allowed: str = "", current: str = PASSWORD) -> Any:
-def lan_client(client: TestClient) -> TestClient:
+    token = (
-    return _other(client, "10.0.10.50")
+        csrf_of(c.get("/settings").text)
-
+        if "X-Test-CSRF" not in c.headers
-
+        else c.headers["X-Test-CSRF"]
-def test_default_requires_login(client: TestClient, lan_client: TestClient) -> None:
+    )
-    assert lan_client.get("/").status_code == 303
+    return c.post(
-
-
-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": trusted,
-            "trusted_networks": "",
+            "allowed_networks": allowed,
-            "current": PASSWORD,
+            "current": current,
         },
     )
-    assert lan_client.get("/").status_code == 303
+
+
+def _set(state: Any, **kw: str) -> None:
+    cfg = load_settings(state.session_factory)
+    for k, v in kw.items():
+        setattr(cfg, k, v)
+    save_settings(state.session_factory, cfg)
+
+
+# -- Entscheidung ------------------------------------------------------------------------------
+
+
+@pytest.mark.parametrize(
+    ("host", "trusted", "allowed", "expected"),
+    [
+        ("10.0.10.50", "", "", ACCESS_PW),  # Standard: überall mit Passwort
+        ("10.0.10.50", "10.0.10.0/24", "", ACCESS_NOPW),
+        ("10.1.5.9", "10.0.10.0/24", "", ACCESS_PW),  # übrige Netze mit Passwort
+        ("10.1.5.9", "10.0.10.0/24", "10.1.5.0/24", ACCESS_PW),
+        ("192.168.5.5", "10.0.10.0/24", "10.1.5.0/24", ACCESS_DENY),
+        ("10.0.10.50", "10.0.10.0/24", "10.1.5.0/24", ACCESS_NOPW),  # ohne PW schlägt Liste
+        ("2001:db8::5", "0.0.0.0/0, ::/0", "", ACCESS_NOPW),  # Anmeldung ganz aus
+        ("testclient", "", "10.0.0.0/8", ACCESS_DENY),  # keine IP-Adresse
+    ],
+)
+def test_access_decision(host: str, trusted: str, allowed: str, expected: str) -> None:
+    assert access_decision(host, "always", trusted, allowed) == expected
+
+
+def test_legacy_off_mode_still_works() -> None:
+    assert access_decision("192.168.5.5", "off", "", "10.1.5.0/24") == ACCESS_NOPW
+
+
+# -- Oberfläche --------------------------------------------------------------------------------
+
+
+def test_default_requires_login(client: TestClient) -> None:
+    assert _other(client, "10.0.10.50").get("/").status_code == 303
+
+
+def test_change_needs_current_password(client: TestClient, state: Any) -> None:
+    _save(client, trusted="10.0.10.0/24", current="falsch")
+    assert load_settings(state.session_factory).trusted_networks == ""
+    assert "Aktuelles Passwort falsch" in client.get("/settings").text
+
+
+def test_without_password_from_trusted_network(client: TestClient, state: Any) -> None:
+    _save(client, trusted="10.0.10.0/24")
+    lan = _other(client, "10.0.10.50")
+    r = lan.get("/")
+    assert r.status_code == 200 and "ohne Anmeldung (vertrautes Netz)" in r.text
+    assert "Abmelden" not in r.text and lan.get("/login").headers["location"] == "/"
+    assert lan.post("/maintenance").status_code == 403  # CSRF gilt weiterhin
+    assert (
+        lan.post("/maintenance", data={"csrf_token": csrf_of(lan.get("/").text)}).status_code == 303
+    )
+    assert _other(client, "10.1.5.9").get("/").status_code == 303  # übrige mit Passwort
+
+
+def test_other_networks_are_rejected_when_password_list_is_set(
+    client: TestClient, state: Any
+) -> None:
+    _save(client, trusted="0.0.0.0/0")  # damit der Test-Client (keine IP) weiterkommt …
+    _set(state, trusted_networks="10.0.10.0/24", allowed_networks="10.1.5.0/24")  # … dann scharf
+    assert _other(client, "10.0.10.50").get("/").status_code == 200
+    assert _other(client, "10.1.5.9").get("/").headers["location"].startswith("/login")
+    blocked = _other(client, "192.168.5.5")
+    for url in ("/", "/login", "/setup", "/mounts"):
+        assert blocked.get(url).status_code == 403, url
+    assert blocked.get("/healthz").status_code == 200  # Überwachung bleibt erreichbar
+    assert blocked.get("/static/app.css").status_code == 200
+
+
+def test_cannot_lock_yourself_out(client: TestClient, state: Any) -> None:
+    lan = _other(client, "10.0.10.50")
+    _save(client, trusted="10.0.10.0/24")
+    # aus 10.0.10.50 eine Liste speichern, in der das eigene Netz fehlt
+    r = _save(lan, trusted="", allowed="10.1.5.0/24")
+    assert r.status_code == 303
+    assert "ausgesperrt" in lan.get("/settings").text
+    cfg = load_settings(state.session_factory)
+    assert cfg.trusted_networks == "10.0.10.0/24" and cfg.allowed_networks == ""
+    # eigenes Netz bei „mit Passwort“ – erlaubt
+    _save(lan, trusted="10.0.10.0/24", allowed="10.1.5.0/24")
+    assert load_settings(state.session_factory).allowed_networks == "10.1.5.0/24"
+
+
+def test_everything_without_password(client: TestClient) -> None:
+    _save(client, trusted="0.0.0.0/0, ::/0")
+    r = _other(client, "192.168.77.1").get("/rules")
+    assert r.status_code == 200 and ">ohne Anmeldung<" in r.text
 
 
 def test_setup_still_required_without_password(raw_client: TestClient, state: Any) -> None:
-    from netmover.settings_store import save_settings
+    _set(state, trusted_networks="0.0.0.0/0")
-
-    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:
+def test_validation_and_loopback_hint(client: TestClient, state: Any) -> None:
-    _set_mode(client, "trusted", "")
+    _save(client, trusted="10.0.10.0/24, quatsch")
-    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"
+    assert load_settings(state.session_factory).trusted_networks == ""
-    _set_mode(client, "trusted", "127.0.0.1")
+    _save(client, trusted="127.0.0.1")
     assert "jede</strong> Anfrage über den Proxy" in client.get("/settings").text
 
 
@@ -102,13 +157,11 @@ 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)
+    _set(state, auth_mode="off", trusted_networks="0.0.0.0/0", allowed_networks="10.1.5.0/24")
-    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"
+    cfg = load_settings(state.session_factory)
+    assert cfg.auth_mode == "always" and cfg.trusted_networks == "" and cfg.allowed_networks == ""

tests/test_copy_mount.py

@@ -0,0 +1,194 @@
+"""0.8.0: Anbindungen kopieren – inklusive Zugangsdaten."""
+
+from __future__ import annotations
+
+from typing import Any
+
+from fastapi.testclient import TestClient
+from sqlalchemy import select
+from sqlalchemy.orm import Session
+
+from netmover.admin import MountInput, copy_name, delete_mount, save_mount
+from netmover.crypto import SecretBox
+from netmover.models import Credential, CredentialKind, Mount, MountType
+from netmover.mounting.fake import FakeMountBackend
+from netmover.mounting.manager import MountManager
+from tests.webutil import post
+
+
+def _cifs(**kw: object) -> MountInput:
+    data: dict[str, object] = {
+        "name": "NAS",
+        "type": MountType.CIFS,
+        "server": "10.0.10.6",
+        "share": "Scans",
+        "username": "scanner",
+        "password": "Pa$$wort",
+    }
+    data.update(kw)
+    return MountInput(**data)  # type: ignore[arg-type]
+
+
+def test_copy_reuses_own_password_as_independent_copy(
+    session: Session, secretbox: SecretBox, manager: MountManager
+) -> None:
+    src, _ = save_mount(session, secretbox, manager, _cifs())
+    assert src is not None and src.credential is not None
+    copy, err = save_mount(
+        session,
+        secretbox,
+        manager,
+        _cifs(name="Archiv", share="Archiv", password=""),
+        copy_from=src,
+    )
+    assert not err and copy is not None and copy.credential is not None
+    assert copy.credential.id != src.credential.id and not copy.credential.managed
+    assert secretbox.decrypt(copy.credential.secret_enc) == "Pa$$wort"
+    assert copy.credential.username == "scanner"
+    # Vorlage löschen – die Kopie behält ihre Zugangsdaten
+    delete_mount(session, manager, src.id)
+    session.flush()
+    assert secretbox.decrypt(copy.credential.secret_enc) == "Pa$$wort"
+
+
+def test_copy_with_new_password_uses_new_one(
+    session: Session, secretbox: SecretBox, manager: MountManager
+) -> None:
+    src, _ = save_mount(session, secretbox, manager, _cifs())
+    copy, _ = save_mount(
+        session, secretbox, manager, _cifs(name="B", share="B", password="anders"), copy_from=src
+    )
+    assert copy is not None and copy.credential is not None
+    assert secretbox.decrypt(copy.credential.secret_enc) == "anders"
+
+
+def test_copy_with_stored_credentials_shares_them(
+    session: Session, secretbox: SecretBox, manager: MountManager
+) -> None:
+    from netmover.credentials import create_password
+
+    cred = create_password(session, secretbox, "NAS-Scanner", "scanner", "x")
+    src, _ = save_mount(
+        session, secretbox, manager, _cifs(username="", password="", cred_id=str(cred.id))
+    )
+    copy, _ = save_mount(
+        session,
+        secretbox,
+        manager,
+        _cifs(name="B", share="B", username="", password="", cred_id=str(cred.id)),
+        copy_from=src,
+    )
+    assert (
+        copy is not None and copy.credential is cred and src is not None and src.credential is cred
+    )
+
+
+def test_copy_sftp_keeps_keypair_and_host_key_only_for_same_server(
+    session: Session, secretbox: SecretBox, manager: MountManager
+) -> None:
+    sftp = {
+        "type": MountType.SFTP,
+        "server": "10.0.10.17",
+        "share": "/consume",
+        "port": 22,
+        "username": "paperless",
+        "auth": CredentialKind.SSH_KEY,
+        "password": "",
+    }
+    src, _ = save_mount(session, secretbox, manager, _cifs(name="Paperless", **sftp))
+    assert src is not None and src.credential is not None
+    src.host_key = "ssh-ed25519 AAAAhost"
+    same, _ = save_mount(
+        session,
+        secretbox,
+        manager,
+        _cifs(name="Paperless 2", **{**sftp, "share": "/consume2"}),
+        copy_from=src,
+    )
+    other, _ = save_mount(
+        session,
+        secretbox,
+        manager,
+        _cifs(name="Anderer", **{**sftp, "server": "10.0.10.99"}),
+        copy_from=src,
+    )
+    assert same is not None and other is not None
+    for m in (same, other):
+        assert m.credential is not None and m.credential.id != src.credential.id
+        assert m.credential.public_key == src.credential.public_key  # gleiches Schlüsselpaar
+        assert m.credential.secret_enc == src.credential.secret_enc
+    assert same.host_key == "ssh-ed25519 AAAAhost" and other.host_key is None
+
+
+def test_copy_name(session: Session, secretbox: SecretBox, manager: MountManager) -> None:
+    save_mount(session, secretbox, manager, _cifs())
+    assert copy_name(session, "NAS") == "NAS (Kopie)"
+    save_mount(session, secretbox, manager, _cifs(name="NAS (Kopie)", share="x"))
+    assert copy_name(session, "NAS") == "NAS (Kopie 2)"
+
+
+# -- Oberfläche --------------------------------------------------------------------------------
+
+
+MOUNT = {
+    "name": "NAS",
+    "type": "cifs",
+    "server": "nas",
+    "share": "scans",
+    "username": "scanner",
+    "password": "Geheim,123",
+    "auto_mount": "on",
+    "enabled": "on",
+}
+
+
+def _mid(state: Any, name: str) -> int:
+    with state.session_factory() as s:
+        mid = s.scalar(select(Mount.id).where(Mount.name == name))
+        assert mid is not None
+        return mid
+
+
+def test_copy_button_form_test_and_save(
+    client: TestClient, state: Any, backend: FakeMountBackend
+) -> None:
+    backend.export("nas", "scans")
+    backend.export("nas", "archiv")
+    post(client, "/mounts/new", MOUNT)
+    src_id = _mid(state, "NAS")
+    assert f'href="/mounts/{src_id}/copy"' in client.get("/mounts").text
+
+    form = client.get(f"/mounts/{src_id}/copy").text
+    assert "Kopie von „NAS“" in form and 'value="NAS (Kopie)"' in form
+    assert 'value="scanner"' in form and "Geheim,123" not in form
+    assert f'name="copy_from" value="{src_id}"' in form
+
+    # Testen mit übernommenem Passwort (Feld leer)
+    copy_form = {
+        **MOUNT,
+        "name": "NAS (Kopie)",
+        "share": "archiv",
+        "password": "",
+        "copy_from": str(src_id),
+    }
+    r = post(client, "/mounts/test", copy_form)
+    assert "Einhängen erfolgreich" in r.text
+    spec = [c for c in backend.calls if c.action == "mount"][-1].spec
+    assert spec is not None and spec.credentials is not None
+    assert spec.credentials.password == "Geheim,123"
+
+    r = post(client, f"/mounts/{src_id}/copy", copy_form)
+    assert r.status_code == 303, r.text
+    with state.session_factory() as s:
+        copy = s.get(Mount, _mid(state, "NAS (Kopie)"))
+        assert copy is not None and copy.share == "archiv" and copy.slug == "nas-kopie"
+        assert copy.credential is not None
+        assert state.secrets.decrypt(copy.credential.secret_enc) == "Geheim,123"
+        assert len(s.scalars(select(Credential)).all()) == 2
+
+
+def test_copy_validation_error_keeps_copy_context(client: TestClient, state: Any) -> None:
+    post(client, "/mounts/new", MOUNT)
+    src_id = _mid(state, "NAS")
+    r = post(client, f"/mounts/{src_id}/copy", {**MOUNT, "name": "NAS", "password": ""})
+    assert r.status_code == 400 and "bereits vergeben" in r.text and "Kopie von „NAS“" in r.text