Florian WildeandClaude Opus 5.5
686542ff6f
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>
278 lines
11 KiB
Python
278 lines
11 KiB
Python
"""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
|