chore(release): 0.2.1

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

CHANGELOG.md

@@ -8,11 +8,42 @@ Die Version steht an genau einer Stelle: `src/netmover/__init__.py`.
 
 ## [Unreleased]
 
+## [0.2.1] – 2026-10-11
+
+### Behoben
+- **Update von 0.1.x auf 0.2.0 scheiterte**: Migration 0002 brach auf einer
+  Datenbank mit Anbindungen ab (`FOREIGN KEY constraint failed` beim Neuaufbau der
+  Tabelle `credentials`). Weil SQLite-Migrationen nicht in einer Transaktion liefen,
+  blieb dabei ein halber Zustand zurück. Migrationen laufen jetzt ohne
+  Fremdschlüsselprüfung, aber mit Integritätsprüfung vor dem Commit und vollständig
+  in einer Transaktion: Ein Abbruch hinterlässt die Datenbank unverändert.
+  Geprüft mit Tests auf verknüpften Daten und an einer Kopie einer echten Datenbank.
+- `netmover-update` zeigt beim Zurückstufen keinen irreführenden Changelog-Auszug mehr.
+
+## [0.2.0] – 2026-10-11
+
+> ⚠️ **Nicht verwenden** – das Update von 0.1.x scheitert an Migration 0002.
+> Stattdessen 0.2.1 installieren. Wer 0.2.0 schon versucht hat: Die Datenbank aus der
+> Sicherung von `netmover-update` zurückspielen und auf 0.2.1 aktualisieren.
+
+### Hinzugefügt
+- **Schlüsselverwaltung** (Seite „Schlüssel“): SSH-Schlüssel erzeugen (Ed25519) oder
+  importieren (Ed25519, ECDSA, RSA ≥ 2048 Bit; OpenSSH/PEM, auch mit Passphrase),
+  Fingerabdruck und `authorized_keys`-Zeile mit SFTP-Beschränkung zum Kopieren,
+  Umbenennen, Löschen nur wenn unbenutzt. Private Schlüssel werden nie angezeigt.
+- SFTP-Anbindungen können einen gespeicherten Schlüssel auswählen; der Benutzername
+  gehört zur Anbindung (Migration 0002: `credentials.managed`, `mounts.username`).
+- **`netmover-update`**: Update aus dem Git-Repo mit Sicherung von Datenbank,
+  Schlüsseln und Einstellungen, Changelog-Auszug, Neustart und Healthcheck;
+  Versionswahl, Warnung beim Zurückstufen.
+- Installation als Git-Klon unter `/opt/netmover/src`; `install.sh` installiert
+  `git` und das Update-Skript.
+
 ### Geändert
-- SFTP-Anbindungen gelten nach erfolgreichem Test als „erreichbar“ (vorher bis
-  zum ersten Regellauf „nicht geprüft“).
 - Klartext-Hinweis bei `mount error(95)` (keine gemeinsame SMB-Version) mit
   Synology-Einstellung; aufgefallen bei der Anbindung des echten NAS.
+- SFTP-Anbindungen gelten nach erfolgreichem Test als „erreichbar“ (vorher bis
+  zum ersten Regellauf „nicht geprüft“).
 
 ## [0.1.1] – 2026-10-10
 
@@ -134,6 +165,8 @@ Erste Version.
 - Löschen von Zugangsdaten, die noch von einem Mount benutzt werden, trennt den
   Mount nicht mehr stillschweigend ab, sondern wird von der Datenbank abgelehnt.
 
-[Unreleased]: https://git.wilde-server.de/florian.wilde/netmover/compare/v0.1.1...HEAD
+[Unreleased]: https://git.wilde-server.de/florian.wilde/netmover/compare/v0.2.1...HEAD
+[0.2.1]: https://git.wilde-server.de/florian.wilde/netmover/compare/v0.2.0...v0.2.1
+[0.2.0]: https://git.wilde-server.de/florian.wilde/netmover/compare/v0.1.1...v0.2.0
 [0.1.1]: https://git.wilde-server.de/florian.wilde/netmover/compare/v0.1.0...v0.1.1
 [0.1.0]: https://git.wilde-server.de/florian.wilde/netmover/releases/tag/v0.1.0

DEVLOG.md

@@ -14,10 +14,40 @@ Zeiten in Ortszeit (Europe/Berlin).
 | 2026-10-10 | 19:18 | 19:34 | 0:16  | 5       | Weboberfläche und Anmeldung            |
 | 2026-10-10 | 19:40 | 19:50 | 0:10  | 6       | Deployment, README, Release 0.1.0      |
 | 2026-10-10 | 21:36 | 22:05 | 0:29  | –       | Praxistest auf VM, Release 0.1.1       |
-| **Summe**  |       |       | **1:35** |      |                                        |
+| 2026-10-10 | 22:45 | 23:00 | 0:15  | –       | Echtes NAS/Paperless anbinden          |
+| 2026-10-10 | 23:31 | 23:58 | 0:27  | –       | Schlüsselverwaltung, Update-Skript, 0.2.x |
+| **Summe**  |       |       | **2:17** |      |                                        |
 
 ---
 
+## 2026-10-10 · 23:31–23:58 · Schlüsselverwaltung, Update-Skript, Releases 0.2.0/0.2.1
+
+**Erledigt**
+- Schlüsselverwaltung (`keys.py`, Seite „Schlüssel“, Auswahl in SFTP-Anbindungen),
+  Migration 0002 (`credentials.managed`, `mounts.username`).
+- `deploy/update.sh` → `netmover-update`: Tags holen, Sicherung (SQLite-Online-
+  Backup + Schlüssel + env), Checkout, `install.sh`, Healthcheck, Rückweg-Anleitung.
+- VM auf Git-Klon `/opt/netmover/src` umgestellt (Repo ist öffentlich lesbar).
+
+**Zwischenfall beim Update der VM auf 0.2.0**
+- Migration 0002 scheiterte: `FOREIGN KEY constraint failed` bei `DROP TABLE
+  credentials` (Alembic-Batch baut die Tabelle neu; `foreign_keys=ON`, und die
+  Anbindungen verweisen darauf). Danach crashte jeder Neustart an der
+  übriggebliebenen `_alembic_tmp_credentials` (DDL lief ohne Transaktion).
+- `netmover-update` hat den Fehler erkannt und die Rückweg-Befehle genannt.
+  Sicherung zurückgespielt, mit `netmover-update v0.1.1` zurückgestuft – in
+  ~1 Minute wieder lauffähig, NAS-Mount blieb durchgehend eingehängt.
+- Ursache, warum die Tests es nicht fanden: Migrationen wurden nur auf **leeren**
+  Datenbanken getestet. Neu: `tests/test_migrations.py` mit verknüpften Daten und
+  mit Abbruch mitten in der Migration (Datenbank muss unverändert bleiben).
+- Fix: eigene Migrationsverbindung (FK aus, `BEGIN IMMEDIATE`, `transactional_ddl`,
+  `foreign_key_check` vor dem Commit). Vor dem Release an einer Kopie der echten
+  Datenbank auf der VM geprüft.
+- Tag v0.2.0 bleibt (veröffentlicht), im Changelog als „nicht verwenden“ markiert.
+
+**Lehre**
+- Jede künftige Migration bekommt einen Test auf einer befüllten Datenbank.
+
 ## 2026-10-10 · 21:36–22:05 · Praxistest auf der Debian-13-VM, Release 0.1.1
 
 **Umgebung**

README.md

@@ -47,15 +47,36 @@ heißt deshalb immer **kopieren, prüfen, dann erst die Quelle löschen**:
 
 ```bash
 apt install -y git
-git clone https://git.wilde-server.de/florian.wilde/netmover.git
+git clone https://git.wilde-server.de/florian.wilde/netmover.git /opt/netmover/src
-cd netmover
+cd /opt/netmover/src && git checkout "$(git tag -l 'v*' --sort=-v:refname | head -n 1)"
-sudo ./deploy/install.sh
+./deploy/install.sh
 ```
 
-Das Skript installiert `python3-venv`, `cifs-utils` und `nfs-common` und richtet
+Das Skript installiert `python3-venv`, `cifs-utils`, `nfs-common` und `git`. Es
-NetMover in `/opt/netmover/venv` ein. Danach installiert es
+richtet NetMover in `/opt/netmover/venv` ein, installiert
-[`deploy/netmover.service`](deploy/netmover.service) und startet den Dienst.
+[`deploy/netmover.service`](deploy/netmover.service) und das Update-Skript
-Ein Update ist derselbe Aufruf nach einem `git pull`.
+`netmover-update` und startet den Dienst.
+
+### Update
+
+```bash
+netmover-update            # auf die neueste Version
+netmover-update --list     # verfügbare Versionen
+netmover-update v0.2.0     # bestimmte Version
+```
+
+`netmover-update` geht so vor:
+1. Neue Versions-Tags aus Gitea holen und die Änderungen laut Changelog anzeigen.
+2. **Datenbank, `secret.key` und Einstellungen nach `/var/backups/netmover/` sichern**
+   (die letzten 10 bleiben).
+3. Die Version auschecken, installieren und den Dienst neu starten. Neue
+   Datenbank-Migrationen laufen dabei automatisch.
+4. `/healthz` prüfen.
+
+Die Mounts bleiben die ganze Zeit eingehängt. Geht etwas schief, nennt das Skript
+die Befehle, um die Sicherung zurückzuspielen und zur vorherigen Version zu wechseln.
+**Zurückstufen** auf eine ältere Version geht nur mit der Datenbank-Sicherung von
+vor dem Update, wenn das Update eine Migration enthielt; das Skript fragt deshalb nach.
 
 Was wo liegt:
 
@@ -149,11 +170,18 @@ Lege in NetMover die *Anbindung → Neue Anbindung* an:
 | Typ | SFTP (nur Ziel) |
 | Server / SSH-Port | IP des LXC / `22` |
 | Freigabe / Pfad | `/opt/paperless_data/consume` |
-| Anmeldeart | SSH-Schlüssel (NetMover erzeugt ihn) |
+| Anmeldeart | SSH-Schlüssel |
+| Schlüssel | einen gespeicherten Schlüssel wählen (siehe unten) oder „neuer Schlüssel nur für diese Anbindung“ |
 | Benutzername | `paperless` (der Benutzer, unter dem Paperless läuft) |
 
-Nach dem Speichern zeigt NetMover den **öffentlichen Schlüssel**. Trage ihn im LXC
+**Schlüsselverwaltung:** Unter *Schlüssel* erzeugst du Ed25519-Schlüssel oder
-ein, und zwar so, dass der Schlüssel **nur SFTP** darf:
+importierst vorhandene (Ed25519, ECDSA, RSA ≥ 2048 Bit; OpenSSH- oder PEM-Format,
+auch mit Passphrase). Ein Schlüssel kann von mehreren Anbindungen genutzt werden; den
+Benutzernamen legt jede Anbindung selbst fest. Private Schlüssel liegen verschlüsselt
+in der Datenbank und werden nie angezeigt. Die Seite zeigt den Fingerabdruck und eine
+fertige `authorized_keys`-Zeile zum Kopieren.
+
+Trage den öffentlichen Schlüssel im LXC so ein, dass er **nur SFTP** darf:
 
 ```bash
 # im Paperless-LXC als root
@@ -389,7 +417,7 @@ Aufbau:
 | `mounting/` | `MountBackend` (Linux/Fake), Mount-Verwaltung, Test, Wiederherstellung, SFTP |
 | `mover/` | Scan, Stabilität, atomares Schreiben, Kollisionen, Buchführung, Regellauf |
 | `runs.py`, `scheduler.py`, `health.py`, `notify.py` | Protokoll, Zeitsteuerung, Gesundheit, Webhook |
-| `admin.py`, `transfer.py`, `web/` | Formulare, Export/Import, Oberfläche |
+| `admin.py`, `keys.py`, `transfer.py`, `web/` | Formulare, Schlüsselverwaltung, Export/Import, Oberfläche |
 
 Die gesamte Mover-Logik ist mit dem `FakeMountBackend` ohne root testbar. SFTP
 wird gegen einen echten SFTP-Server im Testprozess getestet.

compose.yml

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

deploy/install.sh

@@ -1,8 +1,10 @@
 #!/bin/sh
 # NetMover auf Debian 13 installieren bzw. aktualisieren (als root ausführen).
 #
-#   ./deploy/install.sh                 # aus diesem Verzeichnis installieren
+#   git clone https://git.wilde-server.de/florian.wilde/netmover.git /opt/netmover/src
-#   ./deploy/install.sh 'git+https://git.wilde-server.de/florian.wilde/netmover.git@v0.1.0'
+#   /opt/netmover/src/deploy/install.sh
+#
+# Spätere Updates: netmover-update (siehe deploy/update.sh).
 set -eu
 
 SOURCE="${1:-$(cd "$(dirname "$0")/.." && pwd)}"
@@ -16,7 +18,7 @@ fi
 echo "==> Pakete"
 export DEBIAN_FRONTEND=noninteractive
 apt-get update -qq
-apt-get install -y -qq python3 python3-venv cifs-utils nfs-common >/dev/null
+apt-get install -y -qq python3 python3-venv cifs-utils nfs-common git >/dev/null
 
 echo "==> Virtuelle Umgebung in $PREFIX/venv"
 install -d -m 0755 "$PREFIX"
@@ -31,6 +33,9 @@ if [ ! -f /etc/netmover/netmover.env ]; then
 fi
 install -d -m 0755 /mnt/netmover
 
+echo "==> Update-Skript"
+install -m 0755 "$(dirname "$0")/update.sh" /usr/local/sbin/netmover-update
+
 echo "==> systemd"
 install -m 0644 "$(dirname "$0")/netmover.service" /etc/systemd/system/netmover.service
 systemctl daemon-reload

deploy/update.sh

@@ -0,0 +1,153 @@
+#!/bin/sh
+# NetMover aktualisieren (als root). Wird von install.sh als
+# /usr/local/sbin/netmover-update installiert.
+#
+#   netmover-update            # auf die neueste Version (neuester Tag vX.Y.Z)
+#   netmover-update v0.2.0     # auf eine bestimmte Version (auch zurück)
+#   netmover-update --list     # verfügbare Versionen anzeigen
+#
+# Ablauf: Tags holen → Datenbank und Schlüssel sichern → Version auschecken →
+# install.sh (installiert, migriert beim Start, startet den Dienst neu) →
+# Gesundheit prüfen. Die Mounts bleiben dabei eingehängt.
+set -eu
+
+SRC="${NETMOVER_SRC:-/opt/netmover/src}"
+VENV=/opt/netmover/venv
+ENV_FILE=/etc/netmover/netmover.env
+BACKUP_ROOT=/var/backups/netmover
+KEEP_BACKUPS=10
+
+if [ "$(id -u)" -ne 0 ]; then
+    echo "Bitte als root ausführen." >&2
+    exit 1
+fi
+if [ ! -d "$SRC/.git" ]; then
+    echo "Kein Git-Klon unter $SRC. Einmalig einrichten:" >&2
+    echo "  git clone https://git.wilde-server.de/florian.wilde/netmover.git $SRC" >&2
+    exit 1
+fi
+
+# Einstellungen des Dienstes (Port, Datenverzeichnis)
+NETMOVER_PORT=8080
+NETMOVER_DATA_DIR=/var/lib/netmover
+# shellcheck disable=SC1090
+[ -f "$ENV_FILE" ] && . "$ENV_FILE"
+
+cd "$SRC"
+git fetch --tags --quiet origin
+
+if [ "${1:-}" = "--list" ]; then
+    git tag -l 'v*' --sort=-v:refname
+    exit 0
+fi
+
+TARGET="${1:-$(git tag -l 'v*' --sort=-v:refname | head -n 1)}"
+if ! git rev-parse -q --verify "refs/tags/$TARGET" >/dev/null; then
+    echo "Version $TARGET gibt es nicht. Verfügbar:" >&2
+    git tag -l 'v*' --sort=-v:refname | head -n 10 >&2
+    exit 1
+fi
+CURRENT="v$("$VENV/bin/netmover" version 2>/dev/null || echo "?")"
+echo "Installiert: $CURRENT"
+echo "Ziel:        $TARGET"
+if [ "$CURRENT" = "$TARGET" ] && [ "${FORCE:-0}" != "1" ]; then
+    echo "Bereits auf dem gewünschten Stand (FORCE=1 erzwingt eine Neuinstallation)."
+    exit 0
+fi
+
+NEWEST="$(printf '%s\n%s\n' "$CURRENT" "$TARGET" | sort -V | tail -n 1)"
+if [ "$NEWEST" = "$CURRENT" ] && [ "$CURRENT" != "$TARGET" ]; then
+    echo
+    echo "!! Zurückstufen auf eine ältere Version: Wurde die Datenbank inzwischen migriert,"
+    echo "   startet die alte Version nicht. Dann vorher die Sicherung von *vor* dem Update"
+    echo "   zurückspielen (siehe $BACKUP_ROOT, Anleitung im README)."
+    printf "   Trotzdem fortfahren? [j/N] "
+    read -r answer
+    [ "$answer" = "j" ] || [ "$answer" = "J" ] || exit 1
+fi
+
+if [ "$NEWEST" = "$TARGET" ]; then
+echo
+echo "==> Änderungen laut Changelog"
+git show "$TARGET:CHANGELOG.md" | awk -v cur="${CURRENT#v}" '
+    /^## \[/ { if (index($0, "[" cur "]")) exit; show = ($0 !~ /Unreleased/) }
+    show' | head -n 60
+fi
+
+echo
+echo "==> Sicherung"
+STAMP="$(date +%Y%m%d-%H%M%S)"
+BACKUP="$BACKUP_ROOT/$STAMP-$CURRENT"
+install -d -m 0700 "$BACKUP_ROOT" "$BACKUP"
+# Konsistente Kopie auch bei laufendem Dienst (SQLite-Online-Backup)
+"$VENV/bin/python" - "$NETMOVER_DATA_DIR/netmover.db" "$BACKUP/netmover.db" <<'EOF'
+import sqlite3, sys
+src, dst = sqlite3.connect(sys.argv[1]), sqlite3.connect(sys.argv[2])
+with dst:
+    src.backup(dst)
+src.close(); dst.close()
+EOF
+for f in secret.key session.key; do
+    [ -f "$NETMOVER_DATA_DIR/$f" ] && cp -p "$NETMOVER_DATA_DIR/$f" "$BACKUP/"
+done
+[ -f "$ENV_FILE" ] && cp -p "$ENV_FILE" "$BACKUP/"
+chmod 0600 "$BACKUP"/*
+echo "    $BACKUP"
+# alte Sicherungen aufräumen
+find "$BACKUP_ROOT" -mindepth 1 -maxdepth 1 -type d -print | sort \
+    | head -n "-$KEEP_BACKUPS" | xargs -r rm -rf --
+
+echo
+echo "==> Version $TARGET auschecken und installieren"
+git -c advice.detachedHead=false checkout --quiet "$TARGET"
+"$SRC/deploy/install.sh" "$SRC" >/tmp/netmover-update.log 2>&1 || {
+    echo "install.sh ist fehlgeschlagen – Ausgabe: /tmp/netmover-update.log" >&2
+    tail -n 20 /tmp/netmover-update.log >&2
+    FAILED=1
+}
+
+echo
+echo "==> Prüfung"
+OK=0
+if [ "${FAILED:-0}" != "1" ]; then
+    i=0
+    while [ $i -lt 30 ]; do
+        if "$VENV/bin/python" - "$NETMOVER_PORT" <<'EOF'
+import json, sys, urllib.error, urllib.request
+url = f"http://127.0.0.1:{sys.argv[1]}/healthz"
+try:
+    body = urllib.request.urlopen(url, timeout=3).read()
+except urllib.error.HTTPError as exc:  # 503 = läuft, meldet aber Probleme
+    body = exc.read()
+except OSError:
+    sys.exit(1)
+data = json.loads(body)
+print(f"    Gesundheit: {data['status']}")
+for p in data.get("problems", []):
+    print(f"    – {p}")
+EOF
+        then
+            OK=1
+            break
+        fi
+        i=$((i + 1))
+        sleep 1
+    done
+fi
+
+if [ "$OK" = "1" ] && systemctl is-active --quiet netmover; then
+    echo "    Dienst läuft: $("$VENV/bin/netmover" version)"
+    echo
+    echo "Fertig. Sicherung: $BACKUP"
+    exit 0
+fi
+
+echo
+echo "!! Der Dienst läuft nach dem Update nicht richtig." >&2
+echo "   Log:    journalctl -u netmover -n 50" >&2
+echo "   Zurück auf $CURRENT (Datenbank aus der Sicherung, weil sie migriert sein kann):" >&2
+echo "     systemctl stop netmover" >&2
+echo "     cp $BACKUP/netmover.db $NETMOVER_DATA_DIR/netmover.db" >&2
+echo "     rm -f $NETMOVER_DATA_DIR/netmover.db-wal $NETMOVER_DATA_DIR/netmover.db-shm" >&2
+echo "     netmover-update $CURRENT" >&2
+exit 1

pyproject.toml

@@ -33,6 +33,7 @@ dev = [
     "ruff>=0.6",
     "mypy>=1.11",
     "types-paramiko",
+    "shellcheck-py",
 ]
 
 [project.scripts]

src/netmover/__init__.py

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

src/netmover/admin.py

@@ -84,6 +84,9 @@ def _int(
 # -- Anbindungen ------------------------------------------------------------------------
 
 
+NEW_KEY = "new"
+
+
 @dataclass
 class MountInput:
     name: str = ""
@@ -95,6 +98,9 @@ class MountInput:
     password: str = ""  # leer = unverändert
     domain: str = ""
     auth: CredentialKind = CredentialKind.PASSWORD  # nur sftp: Schlüssel oder Passwort
+    # nur sftp + Schlüssel: "new" = eigener Schlüssel nur für diese Anbindung,
+    # sonst ID eines Schlüssels aus der Schlüsselverwaltung
+    key_id: str = NEW_KEY
     uid: int | None = None
     gid: int | None = None
     file_mode: str | None = None
@@ -126,9 +132,10 @@ def mount_input_from_model(m: Mount) -> MountInput:
         server=m.server,
         share=m.share,
         port=m.port,
-        username=c.username if c else "",
+        username=m.username or (c.username if c else ""),
         domain=(c.domain or "") if c else "",
         auth=c.kind if c else CredentialKind.PASSWORD,
+        key_id=str(c.id) if c is not None and c.managed else NEW_KEY,
         uid=m.uid,
         gid=m.gid,
         file_mode=m.file_mode,
@@ -151,7 +158,11 @@ def parse_mount_form(form: Mapping[str, str]) -> tuple[MountInput, Errors]:
         extra_options=(form.get("extra_options") or "").strip(),
         auto_mount=_bool(form, "auto_mount"),
         enabled=_bool(form, "enabled"),
+        key_id=(form.get("key_id") or NEW_KEY).strip(),
     )
+    if data.key_id != NEW_KEY and not data.key_id.isdigit():
+        errors["key_id"] = "Ungültige Schlüsselauswahl."
+        data.key_id = NEW_KEY
     try:
         data.type = MountType(form.get("type", "cifs"))
     except ValueError:
@@ -203,17 +214,27 @@ def parse_mount_form(form: Mapping[str, str]) -> tuple[MountInput, Errors]:
         errors["username"] = "SFTP braucht einen Benutzernamen."
     if data.type is not MountType.SFTP:
         data.auth = CredentialKind.PASSWORD
+    if data.auth is not CredentialKind.SSH_KEY:
+        data.key_id = NEW_KEY
     return data, errors
 
 
 def credential_for_test(
-    data: MountInput, existing: Mount | None, secrets: SecretBox
+    data: MountInput, existing: Mount | None, secrets: SecretBox, s: Session | None = None
 ) -> CredentialData | None:
-    """Zugangsdaten für „Testen“ aus dem Formular – leeres Passwort = gespeichertes."""
+    """Zugangsdaten für „Testen“ aus dem Formular – leeres Passwort = gespeichertes,
+    gewählter Schlüssel aus der Schlüsselverwaltung."""
     if data.type in (MountType.NFS3, MountType.NFS4) or not data.username:
         return None
+    if data.auth is CredentialKind.SSH_KEY and data.key_id != NEW_KEY and s is not None:
+        key = s.get(Credential, int(data.key_id))
+        if key is None or not key.managed:
+            return None
+        return CredentialData(data.auth, data.username, secrets.decrypt(key.secret_enc))
     secret = data.password
     stored = existing.credential if existing is not None else None
+    if stored is not None and stored.managed:
+        stored = None  # eigener Schlüssel gewählt, gespeicherter ist nicht gemeint
     if not secret and stored is not None and stored.kind is data.auth:
         secret = secrets.decrypt(stored.secret_enc)
     if not secret:
@@ -296,20 +317,32 @@ def save_mount(
     if not data.type.is_kernel_mount:
         m.auto_mount = False
 
-    # Zugangsdaten: jede Anbindung pflegt ihre eigenen.
+    # Zugangsdaten: entweder ein gespeicherter Schlüssel aus der Schlüsselverwaltung
-    cred = m.credential
+    # (geteilt, wird hier nie verändert) oder eigene Zugangsdaten nur dieser Anbindung.
+    current = m.credential
+    own = current if current is not None and not current.managed else None
     needs_cred = data.type is MountType.SFTP or (
         data.type is MountType.CIFS and bool(data.username)
     )
+    use_managed = (
+        data.type is MountType.SFTP
+        and data.auth is CredentialKind.SSH_KEY
+        and data.key_id != NEW_KEY
+    )
     if not needs_cred:
-        if cred is not None:
+        m.credential = None
-            m.credential = None
+        m.username = None
-            s.flush()
+    elif use_managed:
-            if not cred.mounts:
+        key = s.get(Credential, int(data.key_id))
-                s.delete(cred)
+        if key is None or not key.managed or key.kind is not CredentialKind.SSH_KEY:
+            s.rollback()
+            return None, {"key_id": "Schlüssel existiert nicht."}
+        m.credential = key
+        m.username = data.username
     else:
-        kind_changed = cred is not None and cred.kind is not data.auth
+        m.username = None
-        if cred is None or kind_changed:
+        cred = own
+        if cred is None or cred.kind is not data.auth:
             if data.auth is CredentialKind.SSH_KEY:
                 private, public = generate_ssh_keypair(f"netmover@{m.slug}")
                 secret, pub = private, public
@@ -320,9 +353,8 @@ def save_mount(
                     return None, errors
                 secret, pub = data.password, None
             if cred is None:
-                cred = Credential(name=f"mount:{m.slug}")
+                cred = Credential(name=f"mount:{m.slug}", managed=False)
                 s.add(cred)
-                m.credential = cred
             cred.kind = data.auth
             cred.secret_enc = secrets.encrypt(secret)
             cred.public_key = pub
@@ -330,6 +362,12 @@ def save_mount(
             cred.secret_enc = secrets.encrypt(data.password)
         cred.username = data.username
         cred.domain = data.domain or None
+        m.credential = cred
+    # Eigene Zugangsdaten, die nicht mehr gebraucht werden, entfernen.
+    if own is not None and m.credential is not own:
+        s.flush()
+        if not own.mounts:
+            s.delete(own)
     try:
         s.flush()
     except IntegrityError:
@@ -353,7 +391,8 @@ def delete_mount(s: Session, mounts: MountManager, mount_id: int) -> None:
     cred = m.credential
     s.delete(m)
     s.flush()
-    if cred is not None and not cred.mounts:
+    # Eigene Zugangsdaten gehen mit; Schlüssel aus der Schlüsselverwaltung bleiben.
+    if cred is not None and not cred.managed and not cred.mounts:
         s.delete(cred)
 
 

src/netmover/db.py

@@ -33,6 +33,35 @@ def make_engine(url: str) -> Engine:
     return engine
 
 
+def make_migration_engine(url: str) -> Engine:
+    """Verbindung nur für Migrationen.
+
+    * Fremdschlüssel AUS: SQLite baut Tabellen bei Änderungen neu auf (Alembic
+      „batch mode“) und löscht dabei die alte Tabelle. Mit eingeschalteter Prüfung
+      scheitert das, sobald andere Tabellen darauf verweisen (Praxisbefund beim
+      Update auf 0.2.0). Die Integrität prüft ``PRAGMA foreign_key_check`` danach.
+    * Echte Transaktion: pysqlite eigene Transaktionssteuerung abnehmen und selbst
+      ``BEGIN`` senden. So wird eine abgebrochene Migration vollständig
+      zurückgerollt, statt einen halben Zustand zu hinterlassen.
+    """
+    engine = create_engine(url, connect_args={"check_same_thread": False})
+    if url.startswith("sqlite"):
+
+        def on_connect(dbapi_conn: Any, _record: Any) -> None:
+            dbapi_conn.isolation_level = None
+            cur = dbapi_conn.cursor()
+            cur.execute("PRAGMA foreign_keys=OFF")  # außerhalb einer Transaktion setzen
+            cur.execute("PRAGMA busy_timeout=5000")
+            cur.close()
+
+        def on_begin(conn: Any) -> None:
+            conn.exec_driver_sql("BEGIN IMMEDIATE")
+
+        event.listen(engine, "connect", on_connect)
+        event.listen(engine, "begin", on_begin)
+    return engine
+
+
 def make_session_factory(engine: Engine) -> sessionmaker[Session]:
     return sessionmaker(bind=engine, expire_on_commit=False)
 

src/netmover/keys.py

@@ -0,0 +1,197 @@
+"""Schlüsselverwaltung: gespeicherte SSH-Schlüssel für SFTP-Anbindungen.
+
+* Erzeugen (Ed25519) oder Importieren (Ed25519, ECDSA, RSA ≥ 2048 Bit; OpenSSH-
+  oder PEM-Format, optional mit Passphrase). Eine Passphrase wird beim Import
+  einmal benutzt; gespeichert wird der Schlüssel ohne Passphrase, aber mit
+  Fernet verschlüsselt – wie alle Geheimnisse.
+* Der private Schlüssel verlässt NetMover nie: Die Oberfläche zeigt nur den
+  öffentlichen Schlüssel und den Fingerabdruck.
+* Ein Schlüssel kann von mehreren Anbindungen genutzt werden; den Benutzernamen
+  legt jede Anbindung selbst fest. Löschen geht nur, solange ihn keine nutzt.
+"""
+
+from __future__ import annotations
+
+import base64
+import hashlib
+from dataclasses import dataclass, field
+from datetime import datetime
+
+from cryptography.hazmat.primitives import serialization
+from cryptography.hazmat.primitives.asymmetric import ec, ed25519, rsa
+from sqlalchemy import select
+from sqlalchemy.orm import Session
+
+from netmover.admin import AdminError
+from netmover.crypto import SecretBox
+from netmover.models import Credential, CredentialKind, Mount
+
+MAX_KEY_BYTES = 16 * 1024
+MIN_RSA_BITS = 2048
+SFTP_ONLY_PREFIX = 'restrict,command="/usr/lib/openssh/sftp-server" '
+
+
+@dataclass
+class KeyInfo:
+    id: int
+    name: str
+    key_type: str
+    fingerprint: str
+    public_key: str
+    created_at: datetime
+    used_by: list[str] = field(default_factory=list)
+
+    @property
+    def authorized_keys_line(self) -> str:
+        """Zeile für ~/.ssh/authorized_keys, auf SFTP beschränkt."""
+        return SFTP_ONLY_PREFIX + self.public_key
+
+
+def fingerprint(public_key: str) -> str:
+    """SHA256-Fingerabdruck wie ``ssh-keygen -lf``."""
+    blob = base64.b64decode(public_key.split()[1])
+    return "SHA256:" + base64.b64encode(hashlib.sha256(blob).digest()).decode().rstrip("=")
+
+
+def _validate_name(s: Session, name: str, exclude_id: int | None = None) -> str:
+    name = name.strip()
+    if not name:
+        raise AdminError("Name fehlt.")
+    if len(name) > 100:
+        raise AdminError("Name höchstens 100 Zeichen.")
+    if name.startswith("mount:"):
+        raise AdminError("Namen mit „mount:“ sind für NetMover reserviert.")
+    dup = s.scalar(select(Credential.id).where(Credential.name == name))
+    if dup is not None and dup != exclude_id:
+        raise AdminError(f"Ein Schlüssel „{name}“ existiert bereits.")
+    return name
+
+
+def _comment(name: str) -> str:
+    return "netmover:" + "".join(c if c.isalnum() or c in "-_." else "-" for c in name)
+
+
+PrivateKey = ed25519.Ed25519PrivateKey | ec.EllipticCurvePrivateKey | rsa.RSAPrivateKey
+
+
+def _store(s: Session, secrets: SecretBox, name: str, key: PrivateKey) -> Credential:
+    private = key.private_bytes(
+        serialization.Encoding.PEM,
+        serialization.PrivateFormat.OpenSSH,
+        serialization.NoEncryption(),
+    ).decode()
+    public = (
+        key.public_key()
+        .public_bytes(serialization.Encoding.OpenSSH, serialization.PublicFormat.OpenSSH)
+        .decode()
+    )
+    cred = Credential(
+        name=name,
+        kind=CredentialKind.SSH_KEY,
+        username="",
+        secret_enc=secrets.encrypt(private),
+        public_key=f"{public} {_comment(name)}",
+        managed=True,
+    )
+    s.add(cred)
+    s.flush()
+    return cred
+
+
+def generate_key(s: Session, secrets: SecretBox, name: str) -> Credential:
+    return _store(s, secrets, _validate_name(s, name), ed25519.Ed25519PrivateKey.generate())
+
+
+def import_key(
+    s: Session, secrets: SecretBox, name: str, private_key: str, passphrase: str = ""
+) -> Credential:
+    name = _validate_name(s, name)
+    data = private_key.strip().encode()
+    if not data:
+        raise AdminError("Privaten Schlüssel einfügen.")
+    if len(data) > MAX_KEY_BYTES:
+        raise AdminError("Schlüssel ist zu groß.")
+    if b"PUBLIC KEY" in data or data.startswith((b"ssh-", b"ecdsa-")):
+        raise AdminError("Das ist ein öffentlicher Schlüssel – benötigt wird der private.")
+    password = passphrase.encode() if passphrase else None
+    loaders = (
+        [serialization.load_ssh_private_key]
+        if b"OPENSSH PRIVATE KEY" in data
+        else [serialization.load_pem_private_key]
+    )
+    try:
+        key = loaders[0](data, password)
+    except TypeError:
+        raise AdminError(
+            "Der Schlüssel ist mit einer Passphrase geschützt – bitte Passphrase angeben."
+            if password is None
+            else "Der Schlüssel hat keine Passphrase – Feld bitte leer lassen."
+        ) from None
+    except ValueError as exc:
+        msg = str(exc).lower()
+        if password is not None or "password" in msg or "decrypt" in msg:
+            # Bei OpenSSH-Schlüsseln meldet cryptography eine falsche Passphrase
+            # als „Corrupt data“ – beides lässt sich nicht unterscheiden.
+            raise AdminError("Passphrase falsch (oder Schlüssel beschädigt).") from None
+        raise AdminError(
+            "Kein lesbarer privater Schlüssel (erwartet OpenSSH- oder PEM-Format)."
+        ) from None
+    except Exception:  # unbekannte Formate/Algorithmen sauber melden
+        raise AdminError("Schlüsselformat wird nicht unterstützt.") from None
+    if not isinstance(
+        key, ed25519.Ed25519PrivateKey | ec.EllipticCurvePrivateKey | rsa.RSAPrivateKey
+    ):
+        raise AdminError("Unterstützt werden Ed25519, ECDSA und RSA.")
+    if isinstance(key, rsa.RSAPrivateKey) and key.key_size < MIN_RSA_BITS:
+        raise AdminError(f"RSA-Schlüssel mit {key.key_size} Bit ist zu schwach (mind. 2048).")
+    return _store(s, secrets, name, key)
+
+
+def _users(s: Session, key_id: int) -> list[str]:
+    # Direkt abfragen statt key.mounts: die Beziehung ist nach dem Löschen von
+    # Anbindungen in derselben Session veraltet (passive_deletes).
+    return sorted(s.scalars(select(Mount.name).where(Mount.cred_id == key_id)))
+
+
+def list_keys(s: Session) -> list[KeyInfo]:
+    keys = s.scalars(
+        select(Credential).where(Credential.managed.is_(True)).order_by(Credential.name)
+    ).all()
+    result = []
+    for k in keys:
+        pub = k.public_key or ""
+        result.append(
+            KeyInfo(
+                id=k.id,
+                name=k.name,
+                key_type=pub.split()[0] if pub else "?",
+                fingerprint=fingerprint(pub) if pub else "?",
+                public_key=pub,
+                created_at=k.created_at,
+                used_by=_users(s, k.id),
+            )
+        )
+    return result
+
+
+def managed_key(s: Session, key_id: int) -> Credential:
+    key = s.get(Credential, key_id)
+    if key is None or not key.managed or key.kind is not CredentialKind.SSH_KEY:
+        raise AdminError("Schlüssel existiert nicht.")
+    return key
+
+
+def rename_key(s: Session, key_id: int, name: str) -> None:
+    key = managed_key(s, key_id)
+    key.name = _validate_name(s, name, exclude_id=key_id)
+
+
+def delete_key(s: Session, key_id: int) -> str:
+    key = managed_key(s, key_id)
+    users = _users(s, key_id)
+    if users:
+        raise AdminError(f"Schlüssel „{key.name}“ wird noch benutzt von: {', '.join(users)}")
+    name = key.name
+    s.delete(key)
+    s.flush()
+    return name

src/netmover/migrations/env.py

@@ -7,7 +7,7 @@ from logging.config import fileConfig
 from alembic import context
 
 from netmover.config import get_settings
-from netmover.db import make_engine
+from netmover.db import make_migration_engine
 from netmover.models import Base
 
 config = context.config
@@ -33,16 +33,27 @@ def run_migrations_offline() -> None:
 
 
 def run_migrations_online() -> None:
-    engine = make_engine(_url())
+    url = _url()
+    engine = make_migration_engine(url)
     with engine.connect() as connection:
         context.configure(
             connection=connection,
             target_metadata=target_metadata,
             render_as_batch=True,
             compare_type=True,
+            # alles oder nichts: auch SQLite kann DDL in einer Transaktion
+            transactional_ddl=True,
         )
         with context.begin_transaction():
             context.run_migrations()
+            if url.startswith("sqlite"):
+                # Fremdschlüssel waren während der Migration aus – vor dem Commit prüfen.
+                bad = connection.exec_driver_sql("PRAGMA foreign_key_check").fetchall()
+                if bad:
+                    raise RuntimeError(
+                        f"Migration verletzt Fremdschlüssel ({len(bad)} Zeilen), "
+                        f"zurückgerollt: {bad[:5]}"
+                    )
     engine.dispose()
 
 

src/netmover/migrations/versions/0002_key_management.py

@@ -0,0 +1,34 @@
+"""key management
+
+Revision ID: 0002
+Revises: 0001
+Create Date: 2026-10-10
+"""
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "0002"
+down_revision: str | None = "0001"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    with op.batch_alter_table("credentials", schema=None) as batch_op:
+        batch_op.add_column(
+            sa.Column("managed", sa.Boolean(), server_default=sa.text("0"), nullable=False)
+        )
+
+    with op.batch_alter_table("mounts", schema=None) as batch_op:
+        batch_op.add_column(sa.Column("username", sa.String(length=255), nullable=True))
+
+
+def downgrade() -> None:
+    with op.batch_alter_table("mounts", schema=None) as batch_op:
+        batch_op.drop_column("username")
+
+    with op.batch_alter_table("credentials", schema=None) as batch_op:
+        batch_op.drop_column("managed")

src/netmover/models.py

@@ -7,6 +7,9 @@ Abweichungen bzw. Ergänzungen gegenüber dem Mindestumfang aus der Spezifikatio
   (beim ersten Test festgehaltener Hostschlüssel, danach verbindlich).
 * ``credentials.kind`` (``password`` | ``ssh_key``) und ``credentials.public_key``.
 * ``credentials.domain`` – Windows-/AD-Freigaben brauchen oft eine Domäne.
+* ``credentials.managed`` + ``mounts.username`` (Migration 0002) –
+  Schlüsselverwaltung: gespeicherte SSH-Schlüssel, die mehrere Anbindungen mit
+  jeweils eigenem Benutzernamen nutzen.
 * ``mounts.state_since`` – für „hängt seit …“ / „gestört seit …“ auf der Statusseite.
 * ``mounts.write_ok`` / ``mounts.root_squash`` / ``mounts.last_test_*`` – Ergebnis
   des letzten Mount-Tests; ``root_squash`` entscheidet, ob ``chown`` auf NFS-Zielen
@@ -38,6 +41,7 @@ from sqlalchemy import (
     TypeDecorator,
     UniqueConstraint,
 )
+from sqlalchemy import false as sa_false
 from sqlalchemy.engine import Dialect
 from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship
 
@@ -166,6 +170,9 @@ class Credential(Base):
     domain: Mapped[str | None] = mapped_column(String(255))
     secret_enc: Mapped[str] = mapped_column(Text)  # Passwort bzw. privater Schlüssel (Fernet)
     public_key: Mapped[str | None] = mapped_column(Text)  # nur ssh_key, OpenSSH-Format
+    # True: in der Schlüsselverwaltung gepflegt, von mehreren Anbindungen nutzbar.
+    # False: gehört genau einer Anbindung und wird mit ihr gelöscht.
+    managed: Mapped[bool] = mapped_column(Boolean, default=False, server_default=sa_false())
     created_at: Mapped[datetime] = mapped_column(UTCDateTime, default=utcnow)
 
     # passive_deletes="all": das ORM soll cred_id nicht still auf NULL setzen,
@@ -183,6 +190,9 @@ class Mount(Base):
     server: Mapped[str] = mapped_column(String(255))
     share: Mapped[str] = mapped_column(String(1024))
     port: Mapped[int | None] = mapped_column(Integer)  # nur sftp
+    # Benutzer für die Anmeldung; überschreibt credentials.username. Nötig, weil ein
+    # gespeicherter SSH-Schlüssel für verschiedene Server/Benutzer gelten kann.
+    username: Mapped[str | None] = mapped_column(String(255))
     host_key: Mapped[str | None] = mapped_column(Text)  # nur sftp, "<typ> <base64>"
     mountpoint: Mapped[str | None] = mapped_column(String(1024), unique=True)  # NULL bei sftp
     cred_id: Mapped[int | None] = mapped_column(ForeignKey("credentials.id", ondelete="RESTRICT"))

src/netmover/mounting/manager.py

@@ -228,7 +228,7 @@ class MountManager:
             if m.credential is not None:
                 c = m.credential
                 cred = CredentialData(
-                    c.kind, c.username, self.secrets.decrypt(c.secret_enc), c.domain
+                    c.kind, m.username or c.username, self.secrets.decrypt(c.secret_enc), c.domain
                 )
             s.expunge(m)
             return m, cred

src/netmover/web/app.py

@@ -45,6 +45,7 @@ from netmover.admin import (
 from netmover.app_state import NetMover
 from netmover.config import Settings, get_settings
 from netmover.db import session_scope
+from netmover.keys import delete_key, generate_key, import_key, list_keys, rename_key
 from netmover.logging_setup import event, get_logger
 from netmover.models import (
     CollisionPolicy,
@@ -392,10 +393,21 @@ async def mounts_page(request: Request) -> Response:
         pubkeys = {
             m.id: m.credential.public_key
             for m in mounts
-            if m.credential is not None and m.credential.public_key
+            if m.credential is not None and m.credential.public_key and not m.credential.managed
+        }
+        key_names = {
+            m.id: m.credential.name
+            for m in mounts
+            if m.credential is not None and m.credential.managed
         }
         return render(
-            request, "mounts.html", active="mounts", mounts=mounts, reports=reports, pubkeys=pubkeys
+            request,
+            "mounts.html",
+            active="mounts",
+            mounts=mounts,
+            reports=reports,
+            pubkeys=pubkeys,
+            key_names=key_names,
         )
 
 
@@ -409,6 +421,15 @@ def _report_data(raw: str | None) -> dict[str, Any]:
     return data
 
 
+def _key_options(request: Request) -> list[tuple[str, str]]:
+    with nm(request).session_factory() as s:
+        keys = list_keys(s)
+    return [("new", "neuer Schlüssel nur für diese Anbindung")] + [
+        (str(k.id), f"{k.name} ({k.key_type.removeprefix('ssh-')}, {k.fingerprint[7:19]}…)")
+        for k in keys
+    ]
+
+
 def _mount_form(
     request: Request,
     data: MountInput,
@@ -416,7 +437,11 @@ def _mount_form(
     mount: Mount | None = None,
     status_code: int = 200,
 ) -> Response:
-    pub = mount.credential.public_key if mount is not None and mount.credential else None
+    pub = (
+        mount.credential.public_key
+        if mount is not None and mount.credential and not mount.credential.managed
+        else None
+    )
     return render(
         request,
         "mount_form.html",
@@ -426,7 +451,10 @@ def _mount_form(
         errors=errors,
         mount=mount,
         public_key=pub,
-        has_secret=bool(mount is not None and mount.credential is not None),
+        has_secret=bool(
+            mount is not None and mount.credential is not None and not mount.credential.managed
+        ),
+        key_options=_key_options(request),
     )
 
 
@@ -487,7 +515,7 @@ async def mount_test_form(request: Request) -> Response:
     existing_id = int(form["mount_id"]) if form.get("mount_id", "").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)
+        cred = credential_for_test(data, existing, state.secrets, s)
     cfg = data.config()
     if data.type is MountType.SFTP and existing is not None:
         cfg = dataclasses.replace(cfg, host_key=existing.host_key)
@@ -503,7 +531,7 @@ async def mount_test(request: Request, mount_id: int) -> Response:
         if m is None:
             return HTMLResponse("Anbindung existiert nicht.", status_code=404)
         data = mount_input_from_model(m)
-        cred = credential_for_test(data, m, state.secrets)
+        cred = credential_for_test(data, m, state.secrets, s)
         from netmover.mounting.manager import MountConfig
 
         cfg = MountConfig.from_model(m)
@@ -563,6 +591,74 @@ async def mount_reset_hostkey(request: Request, mount_id: int) -> Response:
     return redirect(f"/mounts#mount-{mount_id}")
 
 
+# -- Schlüsselverwaltung ----------------------------------------------------------------------
+
+
+@protected.get("/keys", response_class=HTMLResponse)
+async def keys_page(request: Request) -> Response:
+    with nm(request).session_factory() as s:
+        return render(request, "keys.html", active="keys", keys=list_keys(s))
+
+
+async def _key_action(request: Request, work: Callable[[Any], str]) -> Response:
+    state = nm(request)
+    try:
+        with session_scope(state.session_factory) as s:
+            flash(request, work(s))
+    except AdminError as exc:
+        flash(request, str(exc), "error")
+    return redirect("/keys")
+
+
+@protected.post("/keys/generate")
+async def key_generate(request: Request) -> Response:
+    form = _form_dict(await request.form())
+    secrets = nm(request).secrets
+
+    def work(s: Any) -> str:
+        key = generate_key(s, secrets, form.get("name", ""))
+        return f"Schlüssel „{key.name}“ erzeugt."
+
+    return await _key_action(request, work)
+
+
+@protected.post("/keys/import")
+async def key_import(request: Request) -> Response:
+    form = _form_dict(await request.form())
+    secrets = nm(request).secrets
+
+    def work(s: Any) -> str:
+        key = import_key(
+            s,
+            secrets,
+            form.get("name", ""),
+            form.get("private_key", ""),
+            form.get("passphrase", ""),
+        )
+        return f"Schlüssel „{key.name}“ importiert."
+
+    return await _key_action(request, work)
+
+
+@protected.post("/keys/{key_id}/rename")
+async def key_rename(request: Request, key_id: int) -> Response:
+    form = _form_dict(await request.form())
+
+    def work(s: Any) -> str:
+        rename_key(s, key_id, form.get("name", ""))
+        return "Schlüssel umbenannt."
+
+    return await _key_action(request, work)
+
+
+@protected.post("/keys/{key_id}/delete")
+async def key_delete(request: Request, key_id: int) -> Response:
+    def work(s: Any) -> str:
+        return f"Schlüssel „{delete_key(s, key_id)}“ gelöscht."
+
+    return await _key_action(request, work)
+
+
 # -- Regeln -----------------------------------------------------------------------------------
 
 

src/netmover/web/templates/base.html

@@ -17,6 +17,7 @@
   <nav class="main">
     <a href="/" class="{{ 'active' if active == 'status' }}">Status</a>
     <a href="/mounts" class="{{ 'active' if active == 'mounts' }}">Anbindungen</a>
+    <a href="/keys" class="{{ 'active' if active == 'keys' }}">Schlüssel</a>
     <a href="/rules" class="{{ 'active' if active == 'rules' }}">Regeln</a>
     <a href="/log" class="{{ 'active' if active == 'log' }}">Protokoll</a>
     <a href="/settings" class="{{ 'active' if active == 'settings' }}">Einstellungen</a>

src/netmover/web/templates/keys.html

@@ -0,0 +1,56 @@
+{% extends "base.html" %}
+{% from "macros.html" import csrf_field %}
+{% block title %}Schlüssel{% endblock %}
+{% block content %}
+<h1>SSH-Schlüssel</h1>
+<p class="muted small">Gespeicherte Schlüssel für SFTP-Anbindungen. Ein Schlüssel kann von mehreren
+  Anbindungen benutzt werden; den Benutzernamen legt jede Anbindung selbst fest. Private Schlüssel
+  liegen verschlüsselt in der Datenbank und werden nie angezeigt.</p>
+
+{% if not keys %}<p class="muted">Noch keine Schlüssel gespeichert.</p>{% endif %}
+<div class="stack">
+{% for k in keys %}
+<div class="card" id="key-{{ k.id }}">
+  <div class="row spread">
+    <div><strong>{{ k.name }}</strong> <span class="badge">{{ k.key_type.removeprefix('ssh-') }}</span>
+      {% if k.used_by %}<span class="badge ok">benutzt von {{ k.used_by|join(', ') }}</span>{% else %}<span class="badge">unbenutzt</span>{% endif %}</div>
+    <div class="row">
+      <form class="inline row" method="post" action="/keys/{{ k.id }}/rename">{{ csrf_field(csrf) }}
+        <input type="text" name="name" value="{{ k.name }}" aria-label="Neuer Name" required maxlength="100">
+        <button class="small">Umbenennen</button></form>
+      <form class="inline" method="post" action="/keys/{{ k.id }}/delete" data-confirm="Schlüssel „{{ k.name }}“ löschen?">{{ csrf_field(csrf) }}
+        <button class="small danger" {{ 'disabled' if k.used_by }}>Löschen</button></form>
+    </div>
+  </div>
+  <div class="small muted">Fingerabdruck <span class="mono">{{ k.fingerprint }}</span> · angelegt {{ k.created_at|localdt }}</div>
+  <details class="small"><summary>Öffentlicher Schlüssel / authorized_keys</summary>
+    <div class="field"><label>Zeile für <code>~/.ssh/authorized_keys</code> (nur SFTP erlaubt, empfohlen)</label>
+      <pre class="mono card" id="auth-{{ k.id }}">{{ k.authorized_keys_line }}</pre>
+      <button class="small" type="button" data-copy="auth-{{ k.id }}">kopieren</button></div>
+    <div class="field"><label>Öffentlicher Schlüssel</label>
+      <pre class="mono card" id="pub-{{ k.id }}">{{ k.public_key }}</pre>
+      <button class="small" type="button" data-copy="pub-{{ k.id }}">kopieren</button></div>
+  </details>
+</div>
+{% endfor %}
+</div>
+
+<h2>Neuen Schlüssel erzeugen</h2>
+<form class="form card row" method="post" action="/keys/generate">{{ csrf_field(csrf) }}
+  <input type="text" name="name" placeholder="Name, z. B. paperless-lxc" aria-label="Name" required maxlength="100">
+  <button class="primary">Ed25519-Schlüssel erzeugen</button>
+</form>
+
+<h2>Vorhandenen Schlüssel importieren</h2>
+<form class="form card stack" method="post" action="/keys/import">{{ csrf_field(csrf) }}
+  <div class="fields">
+    <div class="field"><label for="k-name">Name</label><input id="k-name" type="text" name="name" required maxlength="100"></div>
+    <div class="field"><label for="k-pass">Passphrase (falls vorhanden)</label><input id="k-pass" type="password" name="passphrase" autocomplete="off"></div>
+    <div class="field wide"><label for="k-priv">Privater Schlüssel</label>
+      <textarea id="k-priv" name="private_key" class="mono" rows="8" autocomplete="off" spellcheck="false"
+        placeholder="-----BEGIN OPENSSH PRIVATE KEY-----&#10;…&#10;-----END OPENSSH PRIVATE KEY-----" required></textarea>
+      <span class="hint">Ed25519, ECDSA oder RSA (mind. 2048 Bit), OpenSSH- oder PEM-Format. Eine Passphrase wird nur zum Entschlüsseln benutzt.</span></div>
+  </div>
+  <button>Importieren</button>
+</form>
+{% endblock %}

src/netmover/web/templates/mount_form.html

@@ -23,7 +23,10 @@
   </div></fieldset>
 
   <fieldset data-show-for="type:cifs sftp"><legend>Anmeldung</legend><div class="fields">
-    <div data-show-for="type:sftp">{{ select('auth', 'Anmeldeart', [('ssh_key', 'SSH-Schlüssel (NetMover erzeugt ihn)'), ('password', 'Passwort')], d.auth.value, errors=errors) }}</div>
+    <div data-show-for="type:sftp">{{ select('auth', 'Anmeldeart', [('ssh_key', 'SSH-Schlüssel'), ('password', 'Passwort')], d.auth.value, errors=errors) }}</div>
+    <div data-show-for="type:sftp"><div data-show-for="auth:ssh_key">
+      {{ select('key_id', 'Schlüssel', key_options, d.key_id, errors=errors, hint='Gespeicherte Schlüssel verwalten unter „Schlüssel“.') }}
+    </div></div>
     {{ field('username', 'Benutzername', d.username, errors=errors, attrs='autocomplete="off"', hint='Bei CIFS leer lassen für Gastzugang') }}
     <div data-show-for="type:cifs">{{ field('domain', 'Domäne (optional)', d.domain, errors=errors) }}</div>
     <div class="field{{ ' has-error' if errors.get('password') }}" id="pw-field">
@@ -33,7 +36,7 @@
       <span class="hint">{{ 'Leer lassen = unverändert. ' if has_secret }}Bei SFTP mit Schlüssel leer lassen.</span>
     </div>
   </div>
-  {% if public_key %}<div class="field wide"><label>Öffentlicher Schlüssel (in ~/.ssh/authorized_keys des Zielbenutzers eintragen)</label>
+  {% if public_key %}<div class="field wide"><label>Öffentlicher Schlüssel dieser Anbindung (in ~/.ssh/authorized_keys des Zielbenutzers eintragen)</label>
     <pre class="mono card" id="pubkey">{{ public_key }}</pre><button class="small" type="button" data-copy="pubkey">kopieren</button></div>{% endif %}
   </fieldset>
 

src/netmover/web/templates/mounts.html

@@ -32,6 +32,9 @@
     <div class="small">Hostschlüssel: {% if m.host_key %}<span class="mono">{{ m.host_key.split()[0] }}</span> festgehalten
       <form class="inline" method="post" action="/mounts/{{ m.id }}/reset-hostkey" data-confirm="Hostschlüssel zurücksetzen? Nur tun, wenn der Server neu aufgesetzt wurde.">{{ csrf_field(csrf) }}<button class="small">zurücksetzen</button></form>
       {% else %}<span class="text-warn">noch nicht bestätigt – bitte „Testen“</span>{% endif %}</div>
+    {% if key_names.get(m.id) %}
+    <div class="small">Anmeldung mit gespeichertem Schlüssel <a href="/keys">„{{ key_names[m.id] }}“</a></div>
+    {% endif %}
     {% if pubkeys.get(m.id) %}
     <details class="small"><summary>Öffentlicher Schlüssel für authorized_keys</summary>
       <pre class="mono" id="pub-{{ m.id }}">{{ pubkeys[m.id] }}</pre>

tests/test_keys.py

@@ -0,0 +1,277 @@
+"""Schlüsselverwaltung: erzeugen, importieren, auswählen, schützen."""
+
+from __future__ import annotations
+
+import subprocess
+from collections.abc import Iterator
+from pathlib import Path
+
+import pytest
+from cryptography.hazmat.primitives import serialization
+from cryptography.hazmat.primitives.asymmetric import ec, rsa
+from sqlalchemy import select
+from sqlalchemy.orm import Session, sessionmaker
+
+from netmover.admin import AdminError, MountInput, delete_mount, save_mount
+from netmover.crypto import SecretBox, generate_ssh_keypair
+from netmover.db import session_scope
+from netmover.keys import delete_key, fingerprint, generate_key, import_key, list_keys, rename_key
+from netmover.models import Credential, CredentialKind, Mount, MountType
+from netmover.mounting.manager import MountConfig, MountManager
+from tests.sftp_stub import StubSFTPServer
+
+
+def _sftp_input(**kw: object) -> MountInput:
+    data: dict[str, object] = {
+        "name": "paperless",
+        "type": MountType.SFTP,
+        "server": "127.0.0.1",
+        "share": "/consume",
+        "port": 22,
+        "username": "paperless",
+        "auth": CredentialKind.SSH_KEY,
+    }
+    data.update(kw)
+    return MountInput(**data)  # type: ignore[arg-type]
+
+
+# -- Modul --------------------------------------------------------------------------------------
+
+
+def test_generate_key(session: Session, secretbox: SecretBox) -> None:
+    key = generate_key(session, secretbox, "paperless-lxc")
+    assert key.managed and key.kind is CredentialKind.SSH_KEY
+    assert key.public_key and key.public_key.startswith("ssh-ed25519 ")
+    assert key.public_key.endswith(" netmover:paperless-lxc")
+    assert "PRIVATE" not in key.public_key
+    assert "OPENSSH PRIVATE KEY" in secretbox.decrypt(key.secret_enc)
+    info = list_keys(session)[0]
+    assert info.fingerprint.startswith("SHA256:") and info.used_by == []
+    assert info.authorized_keys_line.startswith(
+        'restrict,command="/usr/lib/openssh/sftp-server" ssh-ed25519 '
+    )
+
+
+def test_fingerprint_matches_ssh_keygen(tmp_path: Path) -> None:
+    _, pub = generate_ssh_keypair("x")
+    f = tmp_path / "k.pub"
+    f.write_text(pub + "\n")
+    try:
+        out = subprocess.run(
+            ["ssh-keygen", "-lf", str(f)],  # noqa: S607 – Abgleich mit OpenSSH
+            capture_output=True,
+            text=True,
+            check=True,
+        ).stdout
+    except FileNotFoundError:
+        pytest.skip("ssh-keygen nicht installiert")
+    assert fingerprint(pub) == out.split()[1]
+
+
+@pytest.mark.parametrize(
+    "kind", ["ed25519-openssh-pass", "rsa-pem", "ecdsa-pem-pass", "rsa-openssh"]
+)
+def test_import_formats(session: Session, secretbox: SecretBox, kind: str) -> None:
+    from cryptography.hazmat.primitives.asymmetric import ed25519
+
+    if kind.startswith("ed25519"):
+        key = ed25519.Ed25519PrivateKey.generate()
+    elif kind.startswith("rsa"):
+        key = rsa.generate_private_key(public_exponent=65537, key_size=2048)  # type: ignore[assignment]
+    else:
+        key = ec.generate_private_key(ec.SECP256R1())  # type: ignore[assignment]
+    enc = (
+        serialization.BestAvailableEncryption(b"geheim")
+        if kind.endswith("pass")
+        else serialization.NoEncryption()
+    )
+    fmt = (
+        serialization.PrivateFormat.OpenSSH
+        if "openssh" in kind
+        else serialization.PrivateFormat.PKCS8
+    )
+    pem = key.private_bytes(serialization.Encoding.PEM, fmt, enc).decode()
+    cred = import_key(session, secretbox, kind, pem, "geheim" if kind.endswith("pass") else "")
+    expected_pub = (
+        key.public_key()
+        .public_bytes(serialization.Encoding.OpenSSH, serialization.PublicFormat.OpenSSH)
+        .decode()
+    )
+    assert cred.public_key is not None and cred.public_key.startswith(expected_pub)
+    stored = secretbox.decrypt(cred.secret_enc)
+    assert "OPENSSH PRIVATE KEY" in stored
+    # ohne Passphrase gespeichert → von paramiko direkt nutzbar
+    from netmover.mounting.sftp import load_private_key
+
+    assert load_private_key(stored).get_base64() == expected_pub.split()[1]
+
+
+def test_import_errors(session: Session, secretbox: SecretBox) -> None:
+    from cryptography.hazmat.primitives.asymmetric import ed25519
+
+    key = ed25519.Ed25519PrivateKey.generate()
+    protected = key.private_bytes(
+        serialization.Encoding.PEM,
+        serialization.PrivateFormat.OpenSSH,
+        serialization.BestAvailableEncryption(b"richtig"),
+    ).decode()
+    with pytest.raises(AdminError, match="Passphrase angeben"):
+        import_key(session, secretbox, "a", protected, "")
+    with pytest.raises(AdminError, match="Passphrase falsch"):
+        import_key(session, secretbox, "a", protected, "falsch")
+    _, pub = generate_ssh_keypair()
+    with pytest.raises(AdminError, match="öffentlicher Schlüssel"):
+        import_key(session, secretbox, "a", pub)
+    with pytest.raises(AdminError, match="Kein lesbarer"):
+        import_key(
+            session, secretbox, "a", "-----BEGIN PRIVATE KEY-----\nMüll\n-----END PRIVATE KEY-----"
+        )
+    weak = (
+        rsa.generate_private_key(public_exponent=65537, key_size=1024)  # noqa: S505 – absichtlich schwach
+        .private_bytes(
+            serialization.Encoding.PEM,
+            serialization.PrivateFormat.PKCS8,
+            serialization.NoEncryption(),
+        )
+        .decode()
+    )
+    with pytest.raises(AdminError, match="zu schwach"):
+        import_key(session, secretbox, "a", weak)
+    assert list_keys(session) == []
+
+
+def test_names(session: Session, secretbox: SecretBox) -> None:
+    generate_key(session, secretbox, "eins")
+    with pytest.raises(AdminError, match="existiert bereits"):
+        generate_key(session, secretbox, "eins")
+    with pytest.raises(AdminError, match="reserviert"):
+        generate_key(session, secretbox, "mount:x")
+    zwei = generate_key(session, secretbox, "zwei")
+    rename_key(session, zwei.id, "drei")
+    assert [k.name for k in list_keys(session)] == ["drei", "eins"]
+    with pytest.raises(AdminError):
+        rename_key(session, zwei.id, "eins")
+
+
+# -- Zusammenspiel mit Anbindungen ----------------------------------------------------------
+
+
+def test_mount_uses_shared_key_with_own_username(
+    session: Session, secretbox: SecretBox, manager: MountManager
+) -> None:
+    key = generate_key(session, secretbox, "lxc")
+    a, err = save_mount(session, secretbox, manager, _sftp_input(key_id=str(key.id)))
+    assert not err and a is not None
+    b, err = save_mount(
+        session,
+        secretbox,
+        manager,
+        _sftp_input(name="anderer", server="10.0.0.9", username="root", key_id=str(key.id)),
+    )
+    assert not err and b is not None
+    assert a.credential is key and b.credential is key
+    assert (a.username, b.username) == ("paperless", "root")
+    assert key.username == ""  # der Schlüssel selbst bleibt unverändert
+    assert list_keys(session)[0].used_by == ["anderer", "paperless"]
+
+    # Schlüssel in Benutzung → nicht löschbar
+    with pytest.raises(AdminError, match="noch benutzt"):
+        delete_key(session, key.id)
+    # Anbindung löschen lässt den gespeicherten Schlüssel stehen
+    delete_mount(session, manager, a.id)
+    delete_mount(session, manager, b.id)
+    assert session.get(Credential, key.id) is not None
+    delete_key(session, key.id)
+    assert session.get(Credential, key.id) is None
+
+
+def test_switching_between_own_and_shared_key(
+    session: Session, secretbox: SecretBox, manager: MountManager
+) -> None:
+    m, _ = save_mount(session, secretbox, manager, _sftp_input())  # eigener Schlüssel
+    assert m is not None and m.credential is not None and not m.credential.managed
+    own_id = m.credential.id
+    key = generate_key(session, secretbox, "geteilt")
+    m, _ = save_mount(session, secretbox, manager, _sftp_input(key_id=str(key.id)), m.id)
+    assert m is not None and m.credential is key
+    assert session.get(Credential, own_id) is None  # eigener Schlüssel aufgeräumt
+    # Wechsel auf Passwort darf den gespeicherten Schlüssel nicht verändern
+    secret_before = key.secret_enc
+    m, _ = save_mount(
+        session,
+        secretbox,
+        manager,
+        _sftp_input(auth=CredentialKind.PASSWORD, password="pw123"),
+        m.id,
+    )
+    assert m is not None and m.credential is not None and m.credential is not key
+    assert not m.credential.managed and m.credential.username == "paperless"
+    assert m.username is None
+    assert key.secret_enc == secret_before and key.kind is CredentialKind.SSH_KEY
+
+
+def test_unknown_or_foreign_key_id_is_rejected(
+    session: Session, secretbox: SecretBox, manager: MountManager
+) -> None:
+    own, _ = save_mount(session, secretbox, manager, _sftp_input(name="a"))
+    assert own is not None and own.credential is not None
+    for bad in ("4711", str(own.credential.id)):  # nicht vorhanden / kein gespeicherter Schlüssel
+        m, err = save_mount(session, secretbox, manager, _sftp_input(name="b", key_id=bad))
+        assert m is None and "key_id" in err
+
+
+@pytest.fixture
+def server(tmp_path: Path) -> Iterator[StubSFTPServer]:
+    root = tmp_path / "sftp-root"
+    (root / "consume").mkdir(parents=True)
+    srv = StubSFTPServer(root, username="paperless").start()
+    yield srv
+    srv.stop()
+
+
+def test_manager_connects_with_shared_key_and_mount_username(
+    server: StubSFTPServer,
+    session_factory: sessionmaker[Session],
+    secretbox: SecretBox,
+    manager: MountManager,
+) -> None:
+    with session_scope(session_factory) as s:
+        key = import_key(s, secretbox, "lxc", server.client_private_key)
+        m, err = save_mount(
+            s, secretbox, manager, _sftp_input(port=server.port, key_id=str(key.id))
+        )
+        assert not err and m is not None
+        mid = m.id
+    with session_factory() as s:
+        mm = s.get(Mount, mid)
+        assert mm is not None
+        cfg = MountConfig.from_model(mm)
+    from netmover.admin import credential_for_test, mount_input_from_model
+
+    with session_factory() as s:
+        mm = s.get(Mount, mid)
+        assert mm is not None
+        data = mount_input_from_model(mm)
+        assert data.key_id == str(mm.cred_id) and data.username == "paperless"
+        cred = credential_for_test(data, mm, secretbox, s)
+    assert cred is not None and cred.username == "paperless"
+    assert manager.check(cfg, cred, mount_id=mid).ok
+    assert manager.check_alive(mid) is None  # Regellauf-Weg: Benutzer der Anbindung
+
+    # Gegenprobe: falscher Benutzer an der Anbindung → Anmeldung abgelehnt
+    with session_scope(session_factory) as s:
+        mm = s.get(Mount, mid)
+        assert mm is not None
+        mm.username = "jemand"
+    from netmover.mounting.manager import MountUnavailable
+
+    with pytest.raises(MountUnavailable, match="Anmeldung abgelehnt"):
+        manager.check_alive(mid)
+
+
+def test_list_keys_ignores_per_mount_credentials(
+    session: Session, secretbox: SecretBox, manager: MountManager
+) -> None:
+    save_mount(session, secretbox, manager, _sftp_input())
+    assert list_keys(session) == []
+    assert session.scalars(select(Credential)).one().managed is False

tests/test_linux_backend.py

@@ -127,7 +127,7 @@ def _alive(pid: int) -> bool:
     try:
         with open(f"/proc/{pid}/stat") as fh:
             return fh.read().split(")")[-1].split()[0] != "Z"
-    except FileNotFoundError:
+    except (FileNotFoundError, ProcessLookupError):  # verschwindet beim Lesen → ESRCH
         return False
 
 

tests/test_migrations.py

@@ -0,0 +1,136 @@
+"""Migrationen auf Datenbanken MIT Daten.
+
+Praxisbefund beim Update 0.1.1 → 0.2.0: Migration 0002 baut auf SQLite Tabellen neu
+auf (batch mode) und löscht dafür die alte Tabelle. Mit eingeschalteter
+Fremdschlüsselprüfung scheiterte das, sobald Anbindungen auf Zugangsdaten
+verwiesen – die Tests liefen bis dahin nur auf leeren Datenbanken. Und weil die
+Migration nicht in einer Transaktion lief, blieb ein halber Zustand zurück.
+"""
+
+from __future__ import annotations
+
+import sqlite3
+from pathlib import Path
+
+import pytest
+from alembic import command
+
+from netmover.db import alembic_config, upgrade
+
+
+def _seed(db: Path) -> None:
+    """Datenbank auf Stand 0001 mit verknüpften Daten, wie auf einem echten System."""
+    c = sqlite3.connect(db)
+    c.execute("PRAGMA foreign_keys=ON")
+    c.executescript(
+        """
+        INSERT INTO credentials (id, name, kind, username, secret_enc, created_at)
+            VALUES (1, 'mount:nas', 'password', 'scanner', 'gAAAA-x', '2026-10-10 20:00:00'),
+                   (2, 'mount:paperless', 'ssh_key', 'root', 'gAAAA-y', '2026-10-10 20:00:00');
+        INSERT INTO mounts (id, name, slug, type, server, share, mountpoint, cred_id,
+                            extra_options, auto_mount, enabled, state)
+            VALUES (1, 'NAS', 'nas', 'cifs', '10.0.10.6', 'Scans', '/mnt/netmover/nas', 1,
+                    '', 1, 1, 'mounted'),
+                   (2, 'paperless', 'paperless', 'sftp', '10.0.10.17', '/consume', NULL, 2,
+                    '', 0, 1, 'mounted');
+        INSERT INTO rules (id, name, enabled, src_mount_id, src_subpath, dst_mount_id, dst_path,
+                           pattern, pattern_type, recursive, interval_seconds, stability_seconds,
+                           min_age_seconds, op_timeout_seconds, action, on_collision, dry_run,
+                           quarantine_after, consecutive_failures, created_at)
+            VALUES (1, 'Scans → Paperless', 0, 1, '', 2, '', '*.pdf', 'glob', 0, 30, 5, 0, 120,
+                    'move', 'rename', 0, 3, 0, '2026-10-10 20:00:00');
+        INSERT INTO runs (id, rule_id, started_at, status, "trigger", dry_run, seen, moved,
+                          skipped, failed)
+            VALUES (1, 1, '2026-10-10 20:00:00', 'ok', 'manual', 1, 3, 3, 0, 0);
+        INSERT INTO run_items (run_id, src, status) VALUES (1, 'a.pdf', 'planned');
+        """
+    )
+    c.commit()
+    c.close()
+
+
+@pytest.fixture
+def db_0001(tmp_path: Path) -> tuple[str, Path]:
+    db = tmp_path / "alt.db"
+    url = f"sqlite:///{db}"
+    upgrade(url, "0001")
+    _seed(db)
+    return url, db
+
+
+def _state(db: Path) -> dict[str, object]:
+    c = sqlite3.connect(db)
+    try:
+        return {
+            "version": c.execute("SELECT version_num FROM alembic_version").fetchone()[0],
+            "tmp_tables": c.execute(
+                "SELECT name FROM sqlite_master WHERE name LIKE '_alembic_tmp%'"
+            ).fetchall(),
+            "cred_columns": [r[1] for r in c.execute("PRAGMA table_info(credentials)")],
+            "mounts": c.execute("SELECT name, cred_id FROM mounts ORDER BY id").fetchall(),
+            "rules": c.execute("SELECT name, src_mount_id, dst_mount_id FROM rules").fetchall(),
+            "run_items": c.execute("SELECT COUNT(*) FROM run_items").fetchone()[0],
+            "fk_violations": c.execute("PRAGMA foreign_key_check").fetchall(),
+        }
+    finally:
+        c.close()
+
+
+def test_upgrade_with_linked_data(db_0001: tuple[str, Path]) -> None:
+    url, db = db_0001
+    upgrade(url)
+    st = _state(db)
+    assert st["version"] == "0002"
+    assert st["tmp_tables"] == [] and st["fk_violations"] == []
+    assert "managed" in st["cred_columns"]  # type: ignore[operator]
+    assert st["mounts"] == [("NAS", 1), ("paperless", 2)]
+    assert st["rules"] == [("Scans → Paperless", 1, 2)]
+    assert st["run_items"] == 1
+    c = sqlite3.connect(db)
+    assert c.execute("SELECT managed FROM credentials ORDER BY id").fetchall() == [(0,), (0,)]
+    assert c.execute("SELECT username FROM mounts ORDER BY id").fetchall() == [(None,), (None,)]
+    c.close()
+
+
+def test_downgrade_with_linked_data(db_0001: tuple[str, Path]) -> None:
+    url, db = db_0001
+    upgrade(url)
+    command.downgrade(alembic_config(url), "0001")
+    st = _state(db)
+    assert st["version"] == "0001" and st["fk_violations"] == []
+    assert st["mounts"] == [("NAS", 1), ("paperless", 2)]
+
+
+def test_failed_migration_leaves_database_untouched(
+    db_0001: tuple[str, Path], monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """Bricht eine Migration mittendrin ab, muss alles beim Alten bleiben – keine
+    halb umgebauten Tabellen, kein _alembic_tmp-Rest, Version unverändert."""
+    from alembic.ddl.impl import DefaultImpl
+
+    url, db = db_0001
+    before = _state(db)
+    real_drop = DefaultImpl.drop_table
+
+    def boom(self: DefaultImpl, *a: object, **kw: object) -> None:
+        raise RuntimeError("Absturz mitten in der Migration")
+
+    monkeypatch.setattr(DefaultImpl, "drop_table", boom)
+    with pytest.raises(RuntimeError, match="Absturz"):
+        upgrade(url)
+    assert _state(db) == before
+
+    monkeypatch.setattr(DefaultImpl, "drop_table", real_drop)
+    upgrade(url)  # danach klappt es ohne Handarbeit
+    assert _state(db)["version"] == "0002"
+
+
+def test_app_engine_still_enforces_foreign_keys(db_url: str) -> None:
+    from sqlalchemy import text
+
+    from netmover.db import make_engine
+
+    engine = make_engine(db_url)
+    with engine.connect() as conn:
+        assert conn.execute(text("PRAGMA foreign_keys")).scalar() == 1
+    engine.dispose()

tests/test_web.py

@@ -154,7 +154,16 @@ def test_security_headers(client: TestClient) -> None:
 
 
 def test_no_inline_scripts_or_styles(client: TestClient) -> None:
-    for url in ("/", "/mounts", "/mounts/new", "/rules", "/rules/new", "/log", "/settings"):
+    for url in (
+        "/",
+        "/mounts",
+        "/mounts/new",
+        "/keys",
+        "/rules",
+        "/rules/new",
+        "/log",
+        "/settings",
+    ):
         html = client.get(url).text
         assert "style=" not in html, url
         assert not re.search(r"<script(?![^>]*\bsrc=)", html), url
@@ -555,3 +564,60 @@ def test_export_excludes_secrets_and_import_roundtrip(
             files={"file": ("export.json", text, "application/json")},
         )
         assert "Übersprungen" in c2.get("/settings").text
+
+
+# -- Schlüsselverwaltung -------------------------------------------------------------------
+
+
+def test_keys_page_generate_import_select_delete(client: TestClient, state: NetMover) -> None:
+    from netmover.crypto import generate_ssh_keypair
+
+    r = post(client, "/keys/generate", {"name": "paperless-lxc"})
+    assert r.status_code == 303
+    private, public = generate_ssh_keypair()
+    r = post(client, "/keys/import", {"name": "alt", "private_key": private, "passphrase": ""})
+    assert r.status_code == 303
+    page = client.get("/keys").text
+    assert "paperless-lxc" in page and "alt" in page
+    assert public.split()[1] in page  # öffentlicher Teil sichtbar
+    for line in private.splitlines()[1:-1]:  # keine Zeile des privaten Schlüssels
+        assert line not in page
+    assert "restrict,command=&#34;/usr/lib/openssh/sftp-server&#34;" in page
+
+    with state.session_factory() as s:
+        key_id = s.scalar(select(Credential.id).where(Credential.name == "alt"))
+    form = client.get("/mounts/new").text
+    assert f'value="{key_id}"' in form and "neuer Schlüssel nur für diese Anbindung" in form
+
+    r = post(
+        client,
+        "/mounts/new",
+        {
+            "name": "paperless",
+            "type": "sftp",
+            "server": "10.0.0.5",
+            "share": "/consume",
+            "port": "22",
+            "auth": "ssh_key",
+            "key_id": str(key_id),
+            "username": "root",
+            "enabled": "on",
+        },
+    )
+    assert r.status_code == 303, r.text
+    mounts = client.get("/mounts").text
+    assert "gespeichertem Schlüssel" in mounts and "„alt“" in mounts
+    edit = client.get(f"/mounts/{_mount_id(state, 'paperless')}/edit").text
+    assert f'<option value="{key_id}" selected>' in edit and 'value="root"' in edit
+
+    post(client, f"/keys/{key_id}/delete")
+    assert "noch benutzt" in client.get("/keys").text
+    post(client, f"/mounts/{_mount_id(state, 'paperless')}/delete")
+    post(client, f"/keys/{key_id}/delete")
+    page = client.get("/keys").text
+    assert "gelöscht" in page and ">alt<" not in page
+
+
+def test_key_import_error_is_shown(client: TestClient) -> None:
+    post(client, "/keys/import", {"name": "x", "private_key": "kein schlüssel", "passphrase": ""})
+    assert "Kein lesbarer privater Schlüssel" in client.get("/keys").text