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>

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

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