chore(release): 0.2.1

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
fix(db): Migrationen ohne Fremdschlüsselprüfung und in einer Transaktion
2026-10-10 23:53:07 +02:00 · 2026-10-10 23:51:15 +02:00 · 2026-10-10 23:43:40 +02:00 · 2026-10-10 23:43:40 +02:00 · 2026-10-10 23:43:40 +02:00 · 2026-10-10 23:43:40 +02:00
24 changed files with 1256 additions and 48 deletions
@@ -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
@@ -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**
@@ -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
-cd netmover
-sudo ./deploy/install.sh
+git clone https://git.wilde-server.de/florian.wilde/netmover.git /opt/netmover/src
+cd /opt/netmover/src && git checkout "$(git tag -l 'v*' --sort=-v:refname | head -n 1)"
+./deploy/install.sh
 ```

-Das Skript installiert `python3-venv`, `cifs-utils` und `nfs-common` und richtet
-NetMover in `/opt/netmover/venv` ein. Danach installiert es
-[`deploy/netmover.service`](deploy/netmover.service) und startet den Dienst.
-Ein Update ist derselbe Aufruf nach einem `git pull`.
+Das Skript installiert `python3-venv`, `cifs-utils`, `nfs-common` und `git`. Es
+richtet NetMover in `/opt/netmover/venv` ein, installiert
+[`deploy/netmover.service`](deploy/netmover.service) und das Update-Skript
+`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
-ein, und zwar so, dass der Schlüssel **nur SFTP** darf:
+**Schlüsselverwaltung:** Unter *Schlüssel* erzeugst du Ed25519-Schlüssel oder
+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.
@@ -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:
@@ -1,8 +1,10 @@
 #!/bin/sh
 # NetMover auf Debian 13 installieren bzw. aktualisieren (als root ausführen).
 #
-#   ./deploy/install.sh                 # aus diesem Verzeichnis installieren
-#   ./deploy/install.sh 'git+https://git.wilde-server.de/florian.wilde/netmover.git@v0.1.0'
+#   git clone https://git.wilde-server.de/florian.wilde/netmover.git /opt/netmover/src
+#   /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
@@ -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
@@ -33,6 +33,7 @@ dev = [
    "ruff>=0.6",
    "mypy>=1.11",
    "types-paramiko",
+    "shellcheck-py",
 ]

 [project.scripts]
@@ -1,3 +1,3 @@
 """NetMover – NFS/SMB-Mounts verwalten und Dateien zeitgesteuert verschieben."""

-__version__ = "0.1.1"
+__version__ = "0.2.1"
@@ -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.
-    cred = m.credential
+    # Zugangsdaten: entweder ein gespeicherter Schlüssel aus der Schlüsselverwaltung
+    # (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
-            s.flush()
-            if not cred.mounts:
-                s.delete(cred)
+        m.credential = None
+        m.username = None
+    elif use_managed:
+        key = s.get(Credential, int(data.key_id))
+        if key is None or not key.managed or key.kind is not CredentialKind.SSH_KEY:
+            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
-        if cred is None or kind_changed:
+        m.username = None
+        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)


@@ -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)

@@ -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
@@ -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()


@@ -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")
@@ -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"))
@@ -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
@@ -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 -----------------------------------------------------------------------------------


@@ -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>
@@ -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 %}
@@ -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>

@@ -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>
@@ -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
@@ -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


@@ -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()
@@ -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
Author	SHA1	Message	Date
Florian WildeandClaude Opus 5.5	a7930c8721	chore(release): 0.2.1 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>	2026-10-10 23:53:07 +02:00
Florian WildeandClaude Opus 5.5	d06bbc6e53	fix(db): Migrationen ohne Fremdschlüsselprüfung und in einer Transaktion Migration 0002 scheiterte auf einer Datenbank mit verknüpften Daten (FOREIGN KEY constraint failed beim Neuaufbau von credentials) und hinterließ einen halben Zustand, weil SQLite-DDL nicht transaktional lief. Eigene Migrationsverbindung: foreign_keys=OFF, BEGIN IMMEDIATE, transactional_ddl, foreign_key_check vor dem Commit. Tests mit echten, verknüpften Daten und mit Abbruch mitten in der Migration. update.sh: beim Zurückstufen keinen Changelog-Auszug. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>	2026-10-10 23:51:15 +02:00
Florian WildeandClaude Opus 5.5	fef90fc059	chore(release): 0.2.0 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>	2026-10-10 23:43:40 +02:00
Florian WildeandClaude Opus 5.5	d46b1ba5c5	test: Wettlauf beim Lesen von /proc/<pid>/stat abfangen Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>	2026-10-10 23:43:40 +02:00
Florian WildeandClaude Opus 5.5	ce5bef8306	feat(deploy): netmover-update mit Sicherung, Changelog-Auszug und Healthcheck Installation als Git-Klon unter /opt/netmover/src; install.sh installiert git und das Update-Skript. Skripte mit shellcheck geprüft. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>	2026-10-10 23:43:40 +02:00
Florian WildeandClaude Opus 5.5	fb975f1eae	feat(web): Seite „Schlüssel“ und Schlüsselauswahl in der Anbindung Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>	2026-10-10 23:43:40 +02:00
Florian WildeandClaude Opus 5.5	686542ff6f	feat: Schlüsselverwaltung für SFTP-Anbindungen Gespeicherte SSH-Schlüssel (erzeugen oder importieren, auch mit Passphrase), von mehreren Anbindungen nutzbar; der Benutzername gehört zur Anbindung. Migration 0002: credentials.managed, mounts.username. Gespeicherte Schlüssel werden beim Löschen einer Anbindung nie mitgelöscht. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>	2026-10-10 23:43:40 +02:00