diff --git a/src/netmover/admin.py b/src/netmover/admin.py index a1e4444..c58d784 100644 --- a/src/netmover/admin.py +++ b/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. - 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) diff --git a/src/netmover/keys.py b/src/netmover/keys.py new file mode 100644 index 0000000..8fc1183 --- /dev/null +++ b/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 diff --git a/src/netmover/migrations/versions/0002_key_management.py b/src/netmover/migrations/versions/0002_key_management.py new file mode 100644 index 0000000..5a41821 --- /dev/null +++ b/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") diff --git a/src/netmover/models.py b/src/netmover/models.py index 1c5c323..4811b5e 100644 --- a/src/netmover/models.py +++ b/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, " " 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")) diff --git a/src/netmover/mounting/manager.py b/src/netmover/mounting/manager.py index 4cfb4b5..64bd8e6 100644 --- a/src/netmover/mounting/manager.py +++ b/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 diff --git a/tests/test_keys.py b/tests/test_keys.py new file mode 100644 index 0000000..10aad8d --- /dev/null +++ b/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