feat: gespeicherte Zugangsdaten für SMB und SFTP-Passwörter
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 <noreply@anthropic.com>
This commit is contained in:
Florian Wilde
co-authored by
Claude Opus 5.5
Claude Opus 5.5
parent
893fe0c95d
commit
f5ea5840d4
@@ -0,0 +1,383 @@
|
||||
"""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.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.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="existieren 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
|
||||
|
||||
|
||||
# -- 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"
|
||||
Reference in New Issue
Block a user