From f5ea5840d4d9c369566c688152658db3319f5b3c Mon Sep 17 00:00:00 2001 From: Florian Wilde Date: Sun, 11 Oct 2026 01:07:09 +0200 Subject: [PATCH] =?UTF-8?q?feat:=20gespeicherte=20Zugangsdaten=20f=C3=BCr?= =?UTF-8?q?=20SMB=20und=20SFTP-Passw=C3=B6rter?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Seite „Zugangsdaten“ (vorher „Schlüssel“) mit Passwörtern und SSH-Schlüsseln; Anbindungen wählen gespeicherte Passwörter aus, ein neues Passwort gilt sofort für alle. /keys leitet auf /credentials um. Co-Authored-By: Claude Opus 5.5 --- src/netmover/admin.py | 50 ++++++++- src/netmover/{keys.py => credentials.py} | 117 +++++++++++++++++++- src/netmover/web/app.py | 110 ++++++++++++++---- src/netmover/web/templates/base.html | 2 +- src/netmover/web/templates/credentials.html | 97 ++++++++++++++++ src/netmover/web/templates/keys.html | 57 ---------- src/netmover/web/templates/mount_form.html | 15 ++- src/netmover/web/templates/mounts.html | 4 +- tests/{test_keys.py => test_credentials.py} | 110 +++++++++++++++++- tests/test_web.py | 62 +++++++++-- 10 files changed, 517 insertions(+), 107 deletions(-) rename src/netmover/{keys.py => credentials.py} (64%) create mode 100644 src/netmover/web/templates/credentials.html delete mode 100644 src/netmover/web/templates/keys.html rename tests/{test_keys.py => test_credentials.py} (70%) diff --git a/src/netmover/admin.py b/src/netmover/admin.py index c4e00a4..d0500c8 100644 --- a/src/netmover/admin.py +++ b/src/netmover/admin.py @@ -85,6 +85,7 @@ def _int( NEW_KEY = "new" +OWN = "own" @dataclass @@ -101,6 +102,8 @@ class MountInput: # 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 + # Passwort-Anmeldung (SMB, SFTP): "own" = eigene Angaben, sonst ID gespeicherter Zugangsdaten + cred_id: str = OWN uid: int | None = None gid: int | None = None file_mode: str | None = None @@ -135,7 +138,12 @@ def mount_input_from_model(m: Mount) -> MountInput: 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, + key_id=str(c.id) + if c is not None and c.managed and c.kind is CredentialKind.SSH_KEY + else NEW_KEY, + cred_id=str(c.id) + if c is not None and c.managed and c.kind is CredentialKind.PASSWORD + else OWN, uid=m.uid, gid=m.gid, file_mode=m.file_mode, @@ -159,10 +167,14 @@ def parse_mount_form(form: Mapping[str, str]) -> tuple[MountInput, Errors]: auto_mount=_bool(form, "auto_mount"), enabled=_bool(form, "enabled"), key_id=(form.get("key_id") or NEW_KEY).strip(), + cred_id=(form.get("cred_id") or OWN).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 + if data.cred_id != OWN and not data.cred_id.isdigit(): + errors["cred_id"] = "Ungültige Auswahl." + data.cred_id = OWN try: data.type = MountType(form.get("type", "cifs")) except ValueError: @@ -210,7 +222,14 @@ def parse_mount_form(form: Mapping[str, str]) -> tuple[MountInput, Errors]: data.username = check("username", lambda: validate_username(data.username), data.username) if data.domain: check("domain", lambda: validate_username(data.domain), "") - if data.type is MountType.SFTP and not data.username: + if data.auth is CredentialKind.SSH_KEY or data.type in (MountType.NFS3, MountType.NFS4): + data.cred_id = OWN + uses_stored = data.cred_id != OWN + if uses_stored: + data.username = data.domain = data.password = "" # kommen aus den Zugangsdaten + errors.pop("username", None) + errors.pop("password", None) + if data.type is MountType.SFTP and not data.username and not uses_stored: errors["username"] = "SFTP braucht einen Benutzernamen." if data.type is not MountType.SFTP: data.auth = CredentialKind.PASSWORD @@ -224,7 +243,19 @@ def credential_for_test( ) -> CredentialData | None: """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: + if data.type in (MountType.NFS3, MountType.NFS4): + return None + if data.cred_id != OWN and s is not None: + stored = s.get(Credential, int(data.cred_id)) + if stored is None or not stored.managed or stored.kind is not CredentialKind.PASSWORD: + return None + return CredentialData( + CredentialKind.PASSWORD, + stored.username, + secrets.decrypt(stored.secret_enc), + stored.domain, + ) + if 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)) @@ -321,8 +352,12 @@ def save_mount( # (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 + stored_password = data.cred_id != OWN and ( + data.type is MountType.CIFS + or (data.type is MountType.SFTP and data.auth is CredentialKind.PASSWORD) + ) needs_cred = data.type is MountType.SFTP or ( - data.type is MountType.CIFS and bool(data.username) + data.type is MountType.CIFS and (bool(data.username) or stored_password) ) use_managed = ( data.type is MountType.SFTP @@ -332,6 +367,13 @@ def save_mount( if not needs_cred: m.credential = None m.username = None + elif stored_password: + stored = s.get(Credential, int(data.cred_id)) + if stored is None or not stored.managed or stored.kind is not CredentialKind.PASSWORD: + s.rollback() + return None, {"cred_id": "Zugangsdaten existieren nicht."} + m.credential = stored + 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: diff --git a/src/netmover/keys.py b/src/netmover/credentials.py similarity index 64% rename from src/netmover/keys.py rename to src/netmover/credentials.py index 8fc1183..f0943ca 100644 --- a/src/netmover/keys.py +++ b/src/netmover/credentials.py @@ -1,4 +1,9 @@ -"""Schlüsselverwaltung: gespeicherte SSH-Schlüssel für SFTP-Anbindungen. +"""Zugangsdaten-Verwaltung: gespeicherte Passwörter und SSH-Schlüssel. + +Passwörter (SMB und SFTP): Benutzer, Passwort, optional Domäne – einmal gepflegt, +von beliebig vielen Anbindungen genutzt. Ein neues Passwort gilt sofort für alle. + +SSH-Schlüssel: * Erzeugen (Ed25519) oder Importieren (Ed25519, ECDSA, RSA ≥ 2048 Bit; OpenSSH- oder PEM-Format, optional mit Passphrase). Eine Passphrase wird beim Import @@ -63,7 +68,7 @@ def _validate_name(s: Session, name: str, exclude_id: int | None = None) -> str: 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.") + raise AdminError(f"Zugangsdaten mit dem Namen „{name}“ existieren bereits.") return name @@ -153,9 +158,115 @@ def _users(s: Session, key_id: int) -> list[str]: return sorted(s.scalars(select(Mount.name).where(Mount.cred_id == key_id))) +# -- Passwörter ------------------------------------------------------------------------------- + + +@dataclass +class PasswordInfo: + id: int + name: str + username: str + domain: str | None + created_at: datetime + used_by: list[str] = field(default_factory=list) + + +def _validate_login(username: str, password: str | None, domain: str) -> None: + from netmover.validation import ValidationError, validate_secret, validate_username + + try: + validate_username(username) + if domain: + validate_username(domain) + if password is not None: + validate_secret(password) + except ValidationError as exc: + raise AdminError(str(exc)) from None + + +def create_password( + s: Session, secrets: SecretBox, name: str, username: str, password: str, domain: str = "" +) -> Credential: + name = _validate_name(s, name) + username, domain = username.strip(), domain.strip() + if not password: + raise AdminError("Passwort fehlt.") + _validate_login(username, password, domain) + cred = Credential( + name=name, + kind=CredentialKind.PASSWORD, + username=username, + domain=domain or None, + secret_enc=secrets.encrypt(password), + managed=True, + ) + s.add(cred) + s.flush() + return cred + + +def update_password( + s: Session, secrets: SecretBox, cred_id: int, username: str, password: str, domain: str = "" +) -> Credential: + """Benutzer/Domäne ändern; leeres Passwort = unverändert.""" + cred = managed_password(s, cred_id) + username, domain = username.strip(), domain.strip() + _validate_login(username, password or None, domain) + cred.username = username + cred.domain = domain or None + if password: + cred.secret_enc = secrets.encrypt(password) + return cred + + +def managed_password(s: Session, cred_id: int) -> Credential: + cred = s.get(Credential, cred_id) + if cred is None or not cred.managed or cred.kind is not CredentialKind.PASSWORD: + raise AdminError("Zugangsdaten existieren nicht.") + return cred + + +def list_passwords(s: Session) -> list[PasswordInfo]: + rows = s.scalars( + select(Credential) + .where(Credential.managed.is_(True), Credential.kind == CredentialKind.PASSWORD) + .order_by(Credential.name) + ).all() + return [ + PasswordInfo(c.id, c.name, c.username, c.domain, c.created_at, _users(s, c.id)) + for c in rows + ] + + +def delete_credential(s: Session, cred_id: int) -> str: + """Gespeicherte Zugangsdaten (Passwort oder Schlüssel) löschen – nur unbenutzt.""" + cred = s.get(Credential, cred_id) + if cred is None or not cred.managed: + raise AdminError("Zugangsdaten existieren nicht.") + users = _users(s, cred_id) + if users: + raise AdminError(f"„{cred.name}“ wird noch benutzt von: {', '.join(users)}") + name = cred.name + s.delete(cred) + s.flush() + return name + + +def rename_credential(s: Session, cred_id: int, name: str) -> None: + cred = s.get(Credential, cred_id) + if cred is None or not cred.managed: + raise AdminError("Zugangsdaten existieren nicht.") + cred.name = _validate_name(s, name, exclude_id=cred_id) + + +# -- SSH-Schlüssel ------------------------------------------------------------------------------ + + def list_keys(s: Session) -> list[KeyInfo]: keys = s.scalars( - select(Credential).where(Credential.managed.is_(True)).order_by(Credential.name) + select(Credential) + .where(Credential.managed.is_(True), Credential.kind == CredentialKind.SSH_KEY) + .order_by(Credential.name) ).all() result = [] for k in keys: diff --git a/src/netmover/web/app.py b/src/netmover/web/app.py index 642e8fe..b7d88ac 100644 --- a/src/netmover/web/app.py +++ b/src/netmover/web/app.py @@ -49,8 +49,17 @@ from netmover.admin import ( ) from netmover.app_state import NetMover from netmover.config import Settings, get_settings +from netmover.credentials import ( + create_password, + delete_credential, + generate_key, + import_key, + list_keys, + list_passwords, + rename_credential, + update_password, +) from netmover.db import session_scope -from netmover.keys import delete_key, generate_key, import_key, list_keys, rename_key from netmover.kuma import push as kuma_push from netmover.logging_setup import event, get_logger from netmover.mail import SECURITY_MODES, MailConfigError, parse_recipients, valid_address @@ -511,7 +520,7 @@ async def mounts_page(request: Request) -> Response: m.id: m.credential.name for m in mounts if m.credential is not None and m.credential.managed - } + } # Name gespeicherter Zugangsdaten (Passwort oder Schlüssel) return render( request, "mounts.html", @@ -533,6 +542,15 @@ def _report_data(raw: str | None) -> dict[str, Any]: return data +def _password_options(request: Request) -> list[tuple[str, str]]: + with nm(request).session_factory() as s: + stored = list_passwords(s) + return [("own", "eigene Angaben für diese Anbindung")] + [ + (str(p.id), f"{p.name} ({p.domain + chr(92) if p.domain else ''}{p.username})") + for p in stored + ] + + def _key_options(request: Request) -> list[tuple[str, str]]: with nm(request).session_factory() as s: keys = list_keys(s) @@ -566,6 +584,7 @@ def _mount_form( mount is not None and mount.credential is not None and not mount.credential.managed ), key_options=_key_options(request), + password_options=_password_options(request), ) @@ -702,26 +721,75 @@ async def mount_reset_hostkey(request: Request, mount_id: int) -> Response: return redirect(f"/mounts#mount-{mount_id}") -# -- Schlüsselverwaltung ---------------------------------------------------------------------- +# -- Zugangsdaten ------------------------------------------------------------------------------ -@protected.get("/keys", response_class=HTMLResponse) -async def keys_page(request: Request) -> Response: +@protected.get("/credentials", response_class=HTMLResponse) +async def credentials_page(request: Request) -> Response: with nm(request).session_factory() as s: - return render(request, "keys.html", active="keys", keys=list_keys(s)) + return render( + request, + "credentials.html", + active="keys", + keys=list_keys(s), + passwords=list_passwords(s), + ) -async def _key_action(request: Request, work: Callable[[Any], str]) -> Response: +@protected.get("/keys") +async def keys_redirect(request: Request) -> Response: + return RedirectResponse("/credentials", status_code=301) + + +async def _cred_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") + return redirect("/credentials") -@protected.post("/keys/generate") +@protected.post("/credentials/passwords") +async def password_create(request: Request) -> Response: + form = _form_dict(await request.form()) + secrets = nm(request).secrets + + def work(s: Any) -> str: + cred = create_password( + s, + secrets, + form.get("name", ""), + form.get("username", ""), + form.get("password", ""), + form.get("domain", ""), + ) + return f"Zugangsdaten „{cred.name}“ gespeichert." + + return await _cred_action(request, work) + + +@protected.post("/credentials/{cred_id}/password") +async def password_update(request: Request, cred_id: int) -> Response: + form = _form_dict(await request.form()) + secrets = nm(request).secrets + + def work(s: Any) -> str: + cred = update_password( + s, + secrets, + cred_id, + form.get("username", ""), + form.get("password", ""), + form.get("domain", ""), + ) + return f"Zugangsdaten „{cred.name}“ geändert – gilt ab sofort für alle Anbindungen." + + return await _cred_action(request, work) + + +@protected.post("/credentials/keys/generate") async def key_generate(request: Request) -> Response: form = _form_dict(await request.form()) secrets = nm(request).secrets @@ -730,10 +798,10 @@ async def key_generate(request: Request) -> Response: key = generate_key(s, secrets, form.get("name", "")) return f"Schlüssel „{key.name}“ erzeugt." - return await _key_action(request, work) + return await _cred_action(request, work) -@protected.post("/keys/import") +@protected.post("/credentials/keys/import") async def key_import(request: Request) -> Response: form = _form_dict(await request.form()) secrets = nm(request).secrets @@ -748,26 +816,26 @@ async def key_import(request: Request) -> Response: ) return f"Schlüssel „{key.name}“ importiert." - return await _key_action(request, work) + return await _cred_action(request, work) -@protected.post("/keys/{key_id}/rename") -async def key_rename(request: Request, key_id: int) -> Response: +@protected.post("/credentials/{cred_id}/rename") +async def credential_rename(request: Request, cred_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." + rename_credential(s, cred_id, form.get("name", "")) + return "Umbenannt." - return await _key_action(request, work) + return await _cred_action(request, work) -@protected.post("/keys/{key_id}/delete") -async def key_delete(request: Request, key_id: int) -> Response: +@protected.post("/credentials/{cred_id}/delete") +async def credential_delete(request: Request, cred_id: int) -> Response: def work(s: Any) -> str: - return f"Schlüssel „{delete_key(s, key_id)}“ gelöscht." + return f"„{delete_credential(s, cred_id)}“ gelöscht." - return await _key_action(request, work) + return await _cred_action(request, work) # -- Regeln ----------------------------------------------------------------------------------- diff --git a/src/netmover/web/templates/base.html b/src/netmover/web/templates/base.html index 32554dd..5de82bd 100644 --- a/src/netmover/web/templates/base.html +++ b/src/netmover/web/templates/base.html @@ -36,7 +36,7 @@
Verbindungen
{{ item('mounts', '/mounts', 'Anbindungen', (nav.mounts_bad ~ ' gestört') if nav and nav.mounts_bad else (nav.mounts if nav and nav.mounts else ''), 'bad' if nav and nav.mounts_bad else '') }} - {{ item('keys', '/keys', 'Schlüssel', nav.keys if nav and nav.keys else '') }} + {{ item('keys', '/credentials', 'Zugangsdaten', nav.keys if nav and nav.keys else '') }}
Automatisierung
diff --git a/src/netmover/web/templates/credentials.html b/src/netmover/web/templates/credentials.html new file mode 100644 index 0000000..548298d --- /dev/null +++ b/src/netmover/web/templates/credentials.html @@ -0,0 +1,97 @@ +{% extends "base.html" %} +{% from "macros.html" import csrf_field %} +{% block title %}Zugangsdaten{% endblock %} +{% block content %} +

Zugangsdaten

+

Einmal hinterlegen, in beliebig vielen Anbindungen auswählen. Passwörter und private Schlüssel + liegen verschlüsselt in der Datenbank und werden nie angezeigt.

+ +

Passwörter ?

+

Für SMB-Freigaben und SFTP-Server mit Passwort. Ein neues Passwort gilt sofort für alle Anbindungen, die es nutzen.

+{% if not passwords %}

Noch keine Passwörter gespeichert.

{% endif %} +
+{% for p in passwords %} +
+
+
{{ p.name }} {% if p.domain %}{{ p.domain }}\{% endif %}{{ p.username }} + {% if p.used_by %}benutzt von {{ p.used_by|join(', ') }}{% else %}unbenutzt{% endif %}
+
+
Ändern +
{{ csrf_field(csrf) }} + + + +
+
Umbenennen +
{{ csrf_field(csrf) }} + +
+
{{ csrf_field(csrf) }} +
+
+
+
angelegt {{ p.created_at|localdt }}
+
+{% endfor %} +
+
{{ csrf_field(csrf) }} +
+
+
+
+
+
+ +
+ +

SSH-Schlüssel

+

Für SFTP-Anbindungen. Ein Schlüssel kann von mehreren Anbindungen benutzt werden; den Benutzernamen legt jede Anbindung selbst fest.

+ +{% if not keys %}

Noch keine Schlüssel gespeichert.

{% endif %} +
+{% for k in keys %} +
+
+
{{ k.name }} {{ k.key_type.removeprefix('ssh-') }} + {% if k.used_by %}benutzt von {{ k.used_by|join(', ') }}{% else %}unbenutzt{% endif %}
+
+
Umbenennen +
{{ csrf_field(csrf) }} + +
+
{{ csrf_field(csrf) }} +
+
+
+
Fingerabdruck {{ k.fingerprint }} · angelegt {{ k.created_at|localdt }}
+
Öffentlicher Schlüssel / authorized_keys +
+ 
{{ k.authorized_keys_line }}
+
+
+ 
{{ k.public_key }}
+
+
+
+{% endfor %} +
+ +

Neuen Schlüssel erzeugen

+
{{ csrf_field(csrf) }} + + +
+ +

Vorhandenen Schlüssel importieren

+
{{ csrf_field(csrf) }} +
+
+
+
+ + Ed25519, ECDSA oder RSA (mind. 2048 Bit), OpenSSH- oder PEM-Format. Eine Passphrase wird nur zum Entschlüsseln benutzt.
+
+ +
+{% endblock %} diff --git a/src/netmover/web/templates/keys.html b/src/netmover/web/templates/keys.html deleted file mode 100644 index 0493ebc..0000000 --- a/src/netmover/web/templates/keys.html +++ /dev/null @@ -1,57 +0,0 @@ -{% extends "base.html" %} -{% from "macros.html" import csrf_field %} -{% block title %}Schlüssel{% endblock %} -{% block content %} -

Schlüssel

-

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.

- -{% if not keys %}

Noch keine Schlüssel gespeichert.

{% endif %} -
-{% for k in keys %} -
-
-
{{ k.name }} {{ k.key_type.removeprefix('ssh-') }} - {% if k.used_by %}benutzt von {{ k.used_by|join(', ') }}{% else %}unbenutzt{% endif %}
-
-
Umbenennen -
{{ csrf_field(csrf) }} - -
-
{{ csrf_field(csrf) }} -
-
-
-
Fingerabdruck {{ k.fingerprint }} · angelegt {{ k.created_at|localdt }}
-
Öffentlicher Schlüssel / authorized_keys -
- 
{{ k.authorized_keys_line }}
-
-
- 
{{ k.public_key }}
-
-
-
-{% endfor %} -
- -

Neuen Schlüssel erzeugen

-
{{ csrf_field(csrf) }} - - -
- -

Vorhandenen Schlüssel importieren

-
{{ csrf_field(csrf) }} -
-
-
-
- - Ed25519, ECDSA oder RSA (mind. 2048 Bit), OpenSSH- oder PEM-Format. Eine Passphrase wird nur zum Entschlüsseln benutzt.
-
- -
-{% endblock %} diff --git a/src/netmover/web/templates/mount_form.html b/src/netmover/web/templates/mount_form.html index 0047664..552af92 100644 --- a/src/netmover/web/templates/mount_form.html +++ b/src/netmover/web/templates/mount_form.html @@ -25,16 +25,21 @@
Anmeldung ?
{{ select('auth', 'Anmeldeart', [('ssh_key', 'SSH-Schlüssel'), ('password', 'Passwort')], d.auth.value, errors=errors) }}
- {{ select('key_id', 'Schlüssel', key_options, d.key_id, errors=errors, hint='Gespeicherte Schlüssel verwalten unter „Schlüssel“.') }} + {{ select('key_id', 'Schlüssel', key_options, d.key_id, errors=errors, hint='Verwalten unter „Zugangsdaten“.') }}
- {{ field('username', 'Benutzername', d.username, errors=errors, attrs='autocomplete="off"') }} -
{{ field('domain', 'Domäne (optional)', d.domain, errors=errors) }}
-
+
+ {{ select('cred_id', 'Zugangsdaten', password_options, d.cred_id, errors=errors, hint='Gespeicherte Passwörter verwalten unter „Zugangsdaten“.') }} +
+
+ {{ field('username', 'Benutzername', d.username, errors=errors, attrs='autocomplete="off"') }} +
+
{{ field('domain', 'Domäne (optional)', d.domain, errors=errors) }}
+
{% if errors.get('password') %}{{ errors['password'] }}{% endif %} {{ 'Leer lassen = unverändert. ' if has_secret }}Benutzer und Passwort leer = Gastzugang. -
+
{% if public_key %}
{{ public_key }}
{% endif %} diff --git a/src/netmover/web/templates/mounts.html b/src/netmover/web/templates/mounts.html index 6a7b49f..481d2f0 100644 --- a/src/netmover/web/templates/mounts.html +++ b/src/netmover/web/templates/mounts.html @@ -28,13 +28,11 @@
{{ m.type.value }} · {% if m.type == MountType.CIFS %}//{{ m.server }}/{{ m.share }}{% elif m.type == MountType.SFTP %}{{ m.server }}:{{ m.port or 22 }}{{ m.share }}{% else %}{{ m.server }}:{{ m.share }}{% endif %} {% if m.mountpoint %} → {{ m.mountpoint }}{% endif %}
{% if m.last_error %}
{{ m.last_error }}
{% endif %} + {% if key_names.get(m.id) %}
Anmeldung mit gespeicherten Zugangsdaten „{{ key_names[m.id] }}“
{% endif %} {% if m.type == MountType.SFTP %}
Hostschlüssel: {% if m.host_key %}{{ m.host_key.split()[0] }} festgehalten
{{ csrf_field(csrf) }}
{% else %}noch nicht bestätigt – bitte „Testen“{% endif %}
- {% if key_names.get(m.id) %} -
Anmeldung mit gespeichertem Schlüssel „{{ key_names[m.id] }}“
- {% endif %} {% if pubkeys.get(m.id) %}
Öffentlicher Schlüssel für authorized_keys 
{{ pubkeys[m.id] }}
diff --git a/tests/test_keys.py b/tests/test_credentials.py similarity index 70% rename from tests/test_keys.py rename to tests/test_credentials.py index 10aad8d..59eb69c 100644 --- a/tests/test_keys.py +++ b/tests/test_credentials.py @@ -13,9 +13,16 @@ from sqlalchemy import select from sqlalchemy.orm import Session, sessionmaker from netmover.admin import AdminError, MountInput, delete_mount, save_mount +from netmover.credentials import ( + delete_key, + fingerprint, + generate_key, + import_key, + list_keys, + rename_key, +) 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 @@ -142,7 +149,7 @@ def test_import_errors(session: Session, secretbox: SecretBox) -> None: def test_names(session: Session, secretbox: SecretBox) -> None: generate_key(session, secretbox, "eins") - with pytest.raises(AdminError, match="existiert bereits"): + with pytest.raises(AdminError, match="existieren bereits"): generate_key(session, secretbox, "eins") with pytest.raises(AdminError, match="reserviert"): generate_key(session, secretbox, "mount:x") @@ -275,3 +282,102 @@ def test_list_keys_ignores_per_mount_credentials( save_mount(session, secretbox, manager, _sftp_input()) assert list_keys(session) == [] assert session.scalars(select(Credential)).one().managed is False + + +# -- Gespeicherte Passwörter (0.5.0) ---------------------------------------------------------- + + +def _cifs(**kw: object) -> MountInput: + data: dict[str, object] = { + "name": "nas", + "type": MountType.CIFS, + "server": "10.0.10.6", + "share": "scans", + } + data.update(kw) + return MountInput(**data) # type: ignore[arg-type] + + +def test_stored_password_shared_by_mounts( + session: Session, secretbox: SecretBox, manager: MountManager +) -> None: + from netmover.credentials import create_password, list_passwords, update_password + + cred = create_password(session, secretbox, "NAS-Scanner", "scanner", "Pa$$,wort", "WG") + a, err = save_mount(session, secretbox, manager, _cifs(cred_id=str(cred.id))) + assert not err and a is not None + b, err = save_mount( + session, secretbox, manager, _cifs(name="archiv", share="archiv", cred_id=str(cred.id)) + ) + assert not err and b is not None + assert a.credential is cred and b.credential is cred and a.username is None + assert list_passwords(session)[0].used_by == ["archiv", "nas"] + + # ein neues Passwort gilt für beide + update_password(session, secretbox, cred.id, "scanner", "neu!", "WG") + from netmover.admin import credential_for_test, mount_input_from_model + + for m in (a, b): + data = mount_input_from_model(m) + assert data.cred_id == str(cred.id) and data.username == "scanner" + c = credential_for_test(data, m, secretbox, session) + assert c is not None and c.secret == "neu!" and c.domain == "WG" + + # Anbindung löschen lässt die gespeicherten Zugangsdaten stehen + delete_mount(session, manager, a.id) + from netmover.credentials import delete_credential + + with pytest.raises(AdminError, match="noch benutzt"): + delete_credential(session, cred.id) + delete_mount(session, manager, b.id) + assert delete_credential(session, cred.id) == "NAS-Scanner" + + +def test_switch_from_own_to_stored_password_cleans_up( + session: Session, secretbox: SecretBox, manager: MountManager +) -> None: + from netmover.credentials import create_password + + m, _ = save_mount(session, secretbox, manager, _cifs(username="alt", password="pw")) + assert m is not None and m.credential is not None + own_id = m.credential.id + cred = create_password(session, secretbox, "geteilt", "scanner", "x") + m, err = save_mount(session, secretbox, manager, _cifs(cred_id=str(cred.id)), m.id) + assert not err and m is not None and m.credential is cred + assert session.get(Credential, own_id) is None + # zurück auf eigene Angaben: gespeicherter Eintrag bleibt unverändert + m, err = save_mount(session, secretbox, manager, _cifs(username="eigen", password="pw2"), m.id) + assert not err and m is not None and not m.credential.managed # type: ignore[union-attr] + assert cred.username == "scanner" and secretbox.decrypt(cred.secret_enc) == "x" + + +def test_stored_password_validation(session: Session, secretbox: SecretBox) -> None: + from netmover.credentials import create_password, update_password + + with pytest.raises(AdminError, match="Passwort fehlt"): + create_password(session, secretbox, "a", "u", "") + with pytest.raises(AdminError): + create_password(session, secretbox, "a", "u", "zeile\numbruch") + cred = create_password(session, secretbox, "a", "u", "pw") + update_password(session, secretbox, cred.id, "u2", "") # leer = unverändert + assert cred.username == "u2" and secretbox.decrypt(cred.secret_enc) == "pw" + + +def test_stored_password_mount_actually_mounts( + session_factory: sessionmaker[Session], + secretbox: SecretBox, + manager: MountManager, + fake_backend: object, +) -> None: + from netmover.credentials import create_password + + fake_backend.export("10.0.10.6", "scans") # type: ignore[attr-defined] + with session_scope(session_factory) as s: + cred = create_password(s, secretbox, "NAS", "scanner", "Pa$$wort", "WG") + m, _ = save_mount(s, secretbox, manager, _cifs(cred_id=str(cred.id))) + assert m is not None + mid = m.id + manager.mount(mid) + spec = fake_backend.calls[-1].spec # type: ignore[attr-defined] + assert spec.credentials.username == "scanner" and spec.credentials.password == "Pa$$wort" + assert spec.credentials.domain == "WG" diff --git a/tests/test_web.py b/tests/test_web.py index 0fb431d..d44ea5f 100644 --- a/tests/test_web.py +++ b/tests/test_web.py @@ -512,12 +512,16 @@ def test_export_excludes_secrets_and_import_roundtrip( 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"}) + r = post(client, "/credentials/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": ""}) + r = post( + client, + "/credentials/keys/import", + {"name": "alt", "private_key": private, "passphrase": ""}, + ) assert r.status_code == 303 - page = client.get("/keys").text + page = client.get("/credentials").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 @@ -546,27 +550,31 @@ def test_keys_page_generate_import_select_delete(client: TestClient, state: NetM ) assert r.status_code == 303, r.text mounts = client.get("/mounts").text - assert "gespeichertem Schlüssel" in mounts and "„alt“" in mounts + assert "gespeicherten Zugangsdaten" in mounts and "„alt“" in mounts edit = client.get(f"/mounts/{_mount_id(state, 'paperless')}/edit").text assert f'